New cloud event API

[sergeuz]

Description

This PR introduces a new API for publishing and subscribing to cloud events. The new API supports sending and receiving up to 16K of data in an event.

Examples

Publishing a text event:

CloudEvent event;
event.name("my_event");
event.data("abc");
Particle.publish(event);
// or
Particle.publish(CloudEvent().name("my_event").data("abc"));

Publishing a binary-encoded event:

CloudEvent event;
event.name("my_event");
event.data("\x01\x02\x03\x04", 4);
event.contentType(ContentType::BINARY);
Particle.publish(event);

Using the event instance as a regular stream:

CloudEvent event;
event.name("my_event");
for (int i = 0; i < 10; ++i) {
    event.write("abc", 3);
}
Particle.publish(event);

Polling the status of an event:

CloudEvent event;
bool done = false;

void loop() {
    if (event.isNew()) {
        event = CloudEvent().name("my_event").data("abc");
        Particle.publish(event);
    }
    if (!done) {
        if (!event.isOk()) {
            // This catches any error that might have occurred while creating or sending the event
            Log.error("Failed to send event: %d", event.error());
            done = true;
        } else if (event.isSent()) {
            Log.info("Successfully sent event");
            done = true;
        }
    }
}

Using a callback to receive status updates for an event:

CloudEvent event;

void loop() {
    if (event.isNew()) {
        event = CloudEvent().name("my_event").data("abc");
        event.onStatusChange([](auto event) {
            if (!event.isOk()) {
                Log.error("Failed to send event: %d", event.error());
            } else if (event.isSent()) {
                Log.info("Successfully sent event");
            }
        });
        Particle.publish(event);
    }
}

Subscribing to events:

void eventHandler(CloudEvent event) {
    Log.info("Received event: %s", event.name());
    Log.info("Event data: %s", event.dataString().c_str());
    // or
    while (event.available() > 0) {
        Serial.print((char)event.read());
    }
}

void setup() {
    Particle.subscribe("my_event", eventHandler);
}

[monkbroc]

Looks good. I mostly looked at the application API for large publishes.

[monkbroc]

todo: make sure that lfs_file_write can handle writing large buffers (>1000 bytes) in one shot or split the writes in here to the limit of lfs_file_write

[sergeuz]

@avtolstoy Is this still something on your team's radar or was it perhaps fixed already? It's that problem with writing to external flash when the source buffer is in internal flash, possibly at an unaligned address.

[avtolstoy]

I have a fix for this, should get a PR out today.

[avtolstoy]

#2843

[sergeuz]

Thanks!

[monkbroc]

Is this still relevant since we only support receiving binary payloads from large events?

[sergeuz]

I mostly added it for feature parity with the existing API. The user can still send CBOR data in a binary event and use this method to parse it. Let me know if we don't want to support encoding and decoding variants in CloudEvent directly and I'll remove the related methods from it.

[sergeuz]

Fixed in 6b4d489.

[monkbroc]

There's no shorthand to see if a CloudEvent is in NEW state. Can you add a new() method?

[sergeuz]

new is a reserved keyword so it will be isNew(). For consistency, I'll need to rename all the other shorthand methods to have an is prefix which I originally omitted for brevity.

[monkbroc]

Or rename the initial state PENDING

[sergeuz]

I added isNew and renamed the other methods to isSending, isOk, etc. PENDING is good but may be confusing for events in progress. Also, incoming events are received in that state too.

[monkbroc]

This one is odd since it checks if the state is not INVALID. Can you make this method invalid() so it's symmetrical with all the other state helpers?

[sergeuz]

I'd prefer to keep it this way so that it's consistent with other APIs unless you feel strongly about it. We only use valid() or isValid() everywhere else in Device OS.

[monkbroc]

ok, sounds good

[monkbroc]

Is this still true? What state does a cancelled event go to?

[sergeuz]

Yeah, a cancelled event gets invalidated. Making it transition to a failed state instead is possible and would be more user friendly but it will require some additional work:

device-os/wiring/src/spark_wiring_cloud_event.cpp

Lines 415 to 417 in 7c89895

	// TODO: For now, transition to an invalid state as an event in a failed state can be sent again
	// and that would create a race condition between cancellation and normal completion of the event
	// in sendComplete()

.

[monkbroc]

OK. No need for additional work here.

[monkbroc]

Is there a way for an application to know how much room is left on the event (or what's the maximum total size of a large event) before calling write and getting COAP_TOO_LARGE_PAYLOAD?

[sergeuz]

The maximum supported size is exposed as CloudEvent::MAX_SIZE. The app can check how much space is left like this:

if (event.pos() + 100 <= CloudEvent::MAX_SIZE) {
    event.write(data, 100);
}

[monkbroc]

So exceeding the amount of data in flight will return LIMIT_EXCEEDED, but the event will stay in the NEW state so it can be sent again later. Is there a way to query how much data is in flight to know when a new event can be sent?

[sergeuz]

the event will stay in the NEW state

That was an oversight. The event will transition to a failed state on this error as well.

Is there a way to query how much data is in flight to know when a new event can be sent?

There's canPublish():

if (CloudEvent::canPublish(3000)) {
    Particle.publish(CloudEvent().name("my_event").data(buf, 3000));
}

I was hesitant to define it as Particle.canPublish() as not to create confusion since we have different limits for the new and old APIs.