Subscribing

Subscribing to data using the SDK involves the following steps:

1. Create an EspSubscriber object. This can be created directly or retrieved from EspProject.
2. Connect the EspSubscriber.
3. Subscribe to streams.
4. In direct access mode, retrieve events using esp_subscriber_get_next_event(). In callback and select access modes, the event is generated by the SDK and passed back to user code.
5. For data events, retrieve EspMessageReader. This encapsulates a single message from the ESP project. It may consist of a single data row or a transaction/envelope block with multiple data rows.
6. Retrieve one or more EspRowReaders. Use the methods in EspRowReader to read in individual fields.

This example shows subscribing to a stream using direct access mode with default options:

EspError * error = esp_error_create();
esp_sdk_start(error);

EspUri * project_uri = esp_uri_create_string("esp://server:port//default/vwap", error);

EspCredentials * creds = esp_credentials_create(ESP_CREDENTIALS_USER_PASSWORD, error);
esp_credentials_set_user(creds, “user”, error);
esp_credentials_set_password(creds, “password”, error);

EspProject * project = esp_project_get(project_uri, creds, NULL, error);

rc = esp_project_connect(project, error);

// Reusing credentials for the subscriber
EspSubscriber * subscriber = esp_project_create_subscriber(project, creds, error);

// Now free credentials
esp_credentials_free(creds, error);

rc = esp_subscriber_connect(subscriber, error);

EspStream * stream = esp_project_get_stream(project, "Trades", error);
rc = esp_subscriber_subsribe(subscriber, stream, error);

while (true) {
  EspSubscriberEvent * event = esp_subscriber_get_next_event(subscriber, error);

  // process event data

  // delete event
  esp_subscriber_event_free(event);
}

esp_subscriber_close(subscriber, error);
esp_sdk_close(); 

If the event is an ESP_SUBSCRIBER_EVENT_DATA event, it contains field data. This is a typical example of reading data from a subscribe event:

    const EspStream * stream = esp_subscriber_event_get_stream(event, error);
          // stream for this event
    EspMessageReader * reader = esp_subscriber_event_get_reader(event, error);
          // get message reader
    int rc = esp_message_reader_is_block(reader, &flag, error);            
          // you can check if this a block
    const EspSchema * schema = esp_stream_get_schema(stream, error);       
          // get the stream schema if you do not have it
    EspRowReader * row_reader;

    int32_t int_value;
    int numcolumns = 0, numrows = 0;
    int type;
    rc = esp_schema_get_numcolumns(schema, &numcolumns, error);         
            // need to know how many columns are there

    while ((row_reader = esp_message_reader_next_row(reader, error)) != NULL) {
            // loop until we finish all rows
        for (int i = 0; i < numcolumns; ++i) {
            rc = esp_row_reader_is_null(row_reader, i, &flag, error);  
            // if column is null, skip
            if ( flag ) 
                continue;
            rc = esp_schema_get_column_type(schema, i, &type, error);
            switch ( type ) {
                case ESP_DATATYPE_INTEGER:
                    rc = esp_row_reader_get_integer(row_reader, i, &int_value, error);
                    break;
                case ESP_DATATYPE_LONG:
                    rc = esp_row_reader_get_long(row_reader, i, &long_value, error);
                    break;
                case ESP_DATATYPE_FLOAT:
                    rc = esp_row_reader_get_float(row_reader, i, &double_value, error);
                // ...
                // other data types
                // ...
            }
        }
    }