CarpDataStreamService

A CARP Core DataStreamService that talks to CAWS.

Uploads collected Measurements to CAWS as DataStreamBatches, and reads them back. Data streams are opened, closed and removed on the server, not by the client.

Key points:

A singleton; call configure (or configureFrom) before use.
appendToDataStreams gzips the payload by default and posts it to the DATA_STREAM_ZIP_ENDPOINT_NAME endpoint.
openDataStreams, closeDataStreams and removeDataStreams always throw a CarpServiceException.
dataStream returns a DataStreamReference for one study deployment.

Apps rarely call it directly: the carp_backend package uploads data through this service when the protocol uses a CAWS data endpoint.

Inheritance

Object
CarpBaseService
CarpDataStreamService

Implemented types

DataStreamService

Constructors

CarpDataStreamService

CarpDataStreamService()

Returns the singleton default instance of the CarpDataStreamService. Before this instance can be used, it must be configured using the configure method.

Constants

DATA_STREAM_ENDPOINT_NAME

String const DATA_STREAM_ENDPOINT_NAME

The default (uncompressed) RPC endpoint name.

DATA_STREAM_QUERY_BY_TIME_ENDPOINT_NAME

String const DATA_STREAM_QUERY_BY_TIME_ENDPOINT_NAME

The endpoint name used by getDataStreamBatchesByTime.

DATA_STREAM_ZIP_ENDPOINT_NAME

String const DATA_STREAM_ZIP_ENDPOINT_NAME

The endpoint name for gzip-compressed uploads.

Properties

rpcEndpointName

@override

String get rpcEndpointName

override

The name of this service's RPC endpoint at CAWS, like deployment-service.

Methods

appendToDataStreams

@override

Future<void> appendToDataStreams(

String studyDeploymentId,
List<DataStreamBatch> batch, {
bool compress = true,

})

override

Appends a batch of data to the data streams of the study deployment with studyDeploymentId.

If compress is true (default), the JSON payload is gzipped before upload.

closeDataStreams

@override

Future<void> closeDataStreams(

List<String> studyDeploymentIds

)

override

Stop accepting incoming data for all data streams for each of the studyDeploymentIds.

Throws IllegalArgumentException when no data streams were ever opened for any of the studyDeploymentIds.

dataStream

DataStreamReference dataStream(

String studyDeploymentId

)

Gets a DataStreamReference for a studyDeploymentId.

getDataStream

@override

Future<List<DataStreamBatch>> getDataStream(

DataStreamId dataStream,
int fromSequenceId, [
int? toSequenceIdInclusive

])

override

Retrieve all data points in dataStream that fall within the inclusive range defined by fromSequenceId and toSequenceIdInclusive. If toSequenceIdInclusive is null, all data points starting fromSequenceId are returned.

In case no data for dataStream is stored in this repository, or is available for the specified range, an empty list is returned.

Throws IllegalArgumentException if:

dataStream has never been opened
the dataStream does not exist (i.e, that the combination of dataStream.deviceRoleName and dataStream.dataType is correct for the protocol used in the dataStream.studyDeploymentId deployment.)
fromSequenceId is negative or toSequenceIdInclusive is smaller than fromSequenceId

getDataStreamBatchesByTime

Future<List<DataStreamBatch>> getDataStreamBatchesByTime(

DataStreamId dataStream,
DateTime from,
DateTime to

)

Query dataStream by its local update time window instead of a sequence-id range.

Returns all data points in dataStream whose local updated_at timestamp falls within the inclusive from-to window, as one DataStreamBatch per contiguous run of measurements - a new batch starts wherever the sequence was interrupted.

This is a CAWS-specific endpoint (not part of the core DataStreamService interface) and mirrors getDataStream, but is useful when the local upload time is more relevant than the sequence id (e.g., incremental sync of recently uploaded data).