olp::dataservice::read::StreamLayerClient Class Reference

Provides the ability to consume data from a stream layer in real time. More...

#include <StreamLayerClient.h>

Public Member Functions
	StreamLayerClient (client::HRN catalog, std::string layer_id, client::OlpClientSettings settings)
	Creates the StreamLayerClient instance. More...
	StreamLayerClient (const StreamLayerClient &other)=delete
	A copy constructor.
	StreamLayerClient (StreamLayerClient &&other) noexcept
	A default move constructor.
StreamLayerClient &	operator= (const StreamLayerClient &other)=delete
	A copy assignment operator.
StreamLayerClient &	operator= (StreamLayerClient &&other) noexcept
	A move assignment operator.
bool	CancelPendingRequests ()
	Cancels all the active and pending requests. More...
client::CancellationToken	Subscribe (SubscribeRequest request, SubscribeResponseCallback callback)
	Enables message consumption for the specific stream layer. More...
client::CancellableFuture< SubscribeResponse >	Subscribe (SubscribeRequest request)
	Enables message consumption for the specific stream layer. More...
client::CancellationToken	Unsubscribe (UnsubscribeResponseCallback callback)
	Deletes the current subscription for the stream layer. More...
client::CancellableFuture< UnsubscribeResponse >	Unsubscribe ()
	Deletes the current subscription for the stream layer. More...
client::CancellationToken	GetData (const model::Message &message, DataResponseCallback callback)
	Downloads the message data using the data handle from the given message metadata. More...
client::CancellableFuture< DataResponse >	GetData (const model::Message &message)
	Downloads message data using a data handle from the given message metadata. More...
client::CancellationToken	Poll (PollResponseCallback callback)
	Reads messages from a stream layer and commits successfully consumed messages before returning them to you. More...
client::CancellableFuture< PollResponse >	Poll ()
	Reads messages from a stream layer and commits successfully consumed messages before returning them to you. More...
client::CancellationToken	Seek (SeekRequest request, SeekResponseCallback callback)
	Allows changing the data stream reading offset. More...
client::CancellableFuture< SeekResponse >	Seek (SeekRequest request)
	Allows changing the data stream reading offset. More...

Detailed Description

Provides the ability to consume data from a stream layer in real time.

The client reads the data in the order it is added to the queue. Once the client reads the data, the data is no longer available to that client, but the data remains available to other clients.

Example of subscribing to and unsubscribing from a stream layer:

auto task_scheduler =
   olp::client::OlpClientSettingsFactory::CreateDefaultTaskScheduler(1u);
auto http_client = olp::client::
   OlpClientSettingsFactory::CreateDefaultNetworkRequestHandler();
 
olp::authentication::Settings settings{"Your.Key.Id", "Your.Key.Secret"};
settings.task_scheduler = task_scheduler;
settings.network_request_handler = http_client;
 
olp::client::AuthenticationSettings auth_settings;
auth_settings.provider =
    olp::authentication::TokenProviderDefault(std::move(settings));
 
auto client_settings = olp::client::OlpClientSettings();
client_settings.authentication_settings = auth_settings;
client_settings.task_scheduler = std::move(task_scheduler);
client_settings.network_request_handler = std::move(http_client);
 
StreamLayerClient client{"hrn:here:data:::your-catalog-hrn", "your-layer-id",
client_settings};
 
auto request =
olp::dataservice::read::SubscribeRequest().WithSubscriptionMode(
    olp::dataservice::read::SubscribeRequest::SubscriptionMode::kSerial));
 
auto subscribe_response = stream_client.Subscribe(request).GetFuture().get();
if (subscribe_response.IsSuccessful()) {
    // Successfully subscribed, now you can consume data from the stream
layer
}
 
// Consume data from the stream layer, seek offsets, and so on.
 
// If you want to stop consumption, you can unsubscribe.
auto unsubscribe_response = stream_client.Unsubscribe().GetFuture().get();
if (unsubscribe_response.IsSuccessful()) {
    // Successfully unsubscribed.
}

See also

The Layers section of the Data User Guide and the Get Data from a Stream Layer section of the Data API Developer Guide.

Constructor & Destructor Documentation

StreamLayerClient()

olp::dataservice::read::StreamLayerClient::StreamLayerClient	(	client::HRN	catalog,
		std::string	layer_id,
		client::OlpClientSettings	settings
	)

Creates the StreamLayerClient instance.

Parameters

catalog	The HRN of the catalog that the stream layer client uses for requests.
layer_id	The ID of the layer that the client uses for requests.
settings	The OlpClientSettings instance.

Member Function Documentation

CancelPendingRequests()

bool olp::dataservice::read::StreamLayerClient::CancelPendingRequests

(

)

Cancels all the active and pending requests.

Returns

True on success.

GetData() [1/2]

client::CancellableFuture<DataResponse> olp::dataservice::read::StreamLayerClient::GetData

(

const model::Message &

message

)

Downloads message data using a data handle from the given message metadata.

Users should use this method to download data only for messages that include a data handle and that have data size greater than 1 MB. Messages with data size less than 1 MB will have the data embedded.

Parameters

message

The Message instance that was retrieved using the Poll method.

Returns

CancellableFuture that contains DataResult or an error. You can also use CancellableFuture to cancel this request.

GetData() [2/2]

client::CancellationToken olp::dataservice::read::StreamLayerClient::GetData	(	const model::Message &	message,
		DataResponseCallback	callback
	)

Downloads the message data using the data handle from the given message metadata.

Users should use this method to download data only for messages that include a data handle and that have data size greater than 1 MB. Messages with data size less than 1 MB will have the data embedded.

Parameters

message	The Message instance that was retrieved using the Poll method.
callback	The DataResponseCallback object that is invoked when the get data request is completed.

Returns

A token that can be used to cancel this request.

Poll() [1/2]

client::CancellableFuture<PollResponse> olp::dataservice::read::StreamLayerClient::Poll

(

)

Reads messages from a stream layer and commits successfully consumed messages before returning them to you.

Only possible if subscribed successfully. If the payload is more than 1 MB, then it is not embedded into the metadata. To download the data, call GetData(Message).

Returns

CancellableFuture that contains PollResponse or an error. You can also use CancellableFuture to cancel this request.

Poll() [2/2]

client::CancellationToken olp::dataservice::read::StreamLayerClient::Poll

(

PollResponseCallback

callback

)

Reads messages from a stream layer and commits successfully consumed messages before returning them to you.

Only possible if subscribed successfully. If the payload is more than 1 MB, then it is not embedded into the metadata. To download the data, call GetData(Message).

Parameters

callback

The PollResponseCallback object that is invoked when the Poll request is completed.

Returns

A token that can be used to cancel this request.

Seek() [1/2]

client::CancellableFuture<SeekResponse> olp::dataservice::read::StreamLayerClient::Seek

(

SeekRequest

request

)

Allows changing the data stream reading offset.

If the commit is successful, the message consumption starts from the specified offset.

Only possible if subscribed successfully. Offsets shouldn't be empty.

Parameters

request

The SeekRequest instance that contains a complete set of request parameters.

Returns

CancellableFuture that contains SeekResponse or an error. You can also use CancellableFuture to cancel this request.

Seek() [2/2]

client::CancellationToken olp::dataservice::read::StreamLayerClient::Seek	(	SeekRequest	request,
		SeekResponseCallback	callback
	)

Allows changing the data stream reading offset.

If the commit is successful, the message consumption starts from the specified offset.

Only possible if subscribed successfully. Offsets shouldn't be empty.

Parameters

request	The SeekRequest instance that contains a complete set of request parameters.
callback	The SeekResponseCallback object that is invoked when the Seek request is completed.

Returns

A token that can be used to cancel this request.

Subscribe() [1/2]

client::CancellableFuture<SubscribeResponse> olp::dataservice::read::StreamLayerClient::Subscribe

(

SubscribeRequest

request

)

Enables message consumption for the specific stream layer.

Parameters

request

The SubscribeRequest instance that contains a complete set of request parameters.

Returns

CancellableFuture that contains the SubscribeId instance or an error. You can also use CancellableFuture to cancel this request.

Subscribe() [2/2]

client::CancellationToken olp::dataservice::read::StreamLayerClient::Subscribe	(	SubscribeRequest	request,
		SubscribeResponseCallback	callback
	)

Enables message consumption for the specific stream layer.

Parameters

request	The SubscribeRequest instance that contains a complete set of request parameters.
callback	The SubscribeResponseCallback object that is invoked when the subscription request is completed.

Returns

A token that can be used to cancel this request.

Unsubscribe() [1/2]

client::CancellableFuture<UnsubscribeResponse> olp::dataservice::read::StreamLayerClient::Unsubscribe

(

)

Deletes the current subscription for the stream layer.

Returns

CancellableFuture that contains the SubscribeId instance of the deleted subscription or an error. You can also use CancellableFuture to cancel this request.

Unsubscribe() [2/2]

client::CancellationToken olp::dataservice::read::StreamLayerClient::Unsubscribe

(

UnsubscribeResponseCallback

callback

)

Deletes the current subscription for the stream layer.

Parameters

callback

The UnsubscribeResponseCallback object that is invoked when the unsubscription request is completed.

Returns

A token that can be used to cancel this request.

---

The documentation for this class was generated from the following file:

• olp-cpp-sdk-dataservice-read/include/olp/dataservice/read/StreamLayerClient.h