com.opendxl.streaming.client

Class Channel

java.lang.Object

com.opendxl.streaming.client.Channel

All Implemented Interfaces:

Consumer, Producer, java.lang.AutoCloseable

public class Channel
extends java.lang.Object
implements Consumer, Producer, java.lang.AutoCloseable

The Channel class is responsible for all communication with the streaming service.

The following example demonstrates the creation of a Channel instance and creating a consumer for the consumer group:

 // Setup consumer properties if non-default configuration values are necessary, e.g.:
 Properties extraConfigs = new Properties();
 extraConfigs.put("enable.auto.commit", false);
 extraConfigs.put("auto.commit.interval.ms", 0);
 extraConfigs.put("auto.offset.reset", "earliest");
 extraConfigs.put("request.timeout.ms", 16000);
 extraConfigs.put("session.timeout.ms", 15000);

 // Setup Http Proxy if required, e.g.:
 HttpProxySettings httpProxySettings = new HttpProxySettings(true, "my-proxy.my-domain.net", 8080, "", "");

 // Create the channel
 Channel channel = new Channel("http://channel-server",       // channelUrl
                               new ChannelAuthUserPass(
                                               "http://channel-server",   // channelUrlLogin
                                               "user",                    // channelUsername
                                               "password",                // channelPassword
                                               null,                      // pathPrefix
                                               "",                        // verifyCertificateBundle
                                               httpProxySettings),        // http proxy settings
                               "thegroup",                    // channelConsumerGroup
                               null,                          // pathPrefix
                               "/databus/consumer-service/v1",  // consumerPathPrefix
                               false,                         // retryOnFail
                               "",                            // verifyCertificateBundle
                               extraConfigs,                  // consumer additional properties
                               httpProxySettings);            // http proxy settings


 // Create a new consumer on the consumer group
 channel.create()
 

Constructor Summary

Constructors

Constructor and Description
Channel(java.lang.String base, ChannelAuth auth, java.lang.String consumerGroup, java.lang.String pathPrefix, java.lang.String consumerPathPrefix, boolean retryOnFail, java.lang.String verifyCertBundle, java.util.Properties extraConfigs, HttpProxySettings httpProxySettings)
Channel(java.lang.String base, ChannelAuth auth, java.lang.String consumerGroup, java.lang.String pathPrefix, java.lang.String consumerPathPrefix, java.lang.String producerPathPrefix, boolean retryOnFail, java.lang.String verifyCertBundle, java.util.Properties extraConfigs, HttpProxySettings httpProxySettings)

Method Summary

Modifier and Type	Method and Description
void	close() Closes the channel.
void	commit() Commits the record offsets to the channel.
ConsumerRecords	consume() Consumes records from all the subscribed topics
ConsumerRecords	consume(int timeout) Consumes records from all the subscribed topics
void	create() Creates a new consumer on the consumer group
void	delete() Deletes the consumer from the consumer group
void	produce(ProducerRecords producerRecords) Produce records to the channel.
void	produce(java.lang.String jsonProducerRecords) Produce records to the channel.
void	run(ConsumerRecordProcessor processCallback, java.util.List<java.lang.String> topics) Repeatedly consume records from the subscribed topics.
void	run(ConsumerRecordProcessor processCallback, java.util.List<java.lang.String> topics, int timeout) Repeatedly consume records from the subscribed topics.
void	run(ConsumerRecordProcessor processCallback, java.lang.String topic) Repeatedly consume records from the subscribed topic.
void	run(ConsumerRecordProcessor processCallback, java.lang.String topic, int timeout) Repeatedly consume records from the subscribed topic.
void	stop() Stop an active execution of the run(ConsumerRecordProcessor, List) call.
void	subscribe(java.util.List<java.lang.String> topics) Subscribes the consumer to a list of topics
java.util.List<java.lang.String>	subscriptions() List the topic names to which the consumer is subscribed.

Methods inherited from class java.lang.Object

equals, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Detail

Channel

public Channel(java.lang.String base,
               ChannelAuth auth,
               java.lang.String consumerGroup,
               java.lang.String pathPrefix,
               java.lang.String consumerPathPrefix,
               boolean retryOnFail,
               java.lang.String verifyCertBundle,
               java.util.Properties extraConfigs,
               HttpProxySettings httpProxySettings)
        throws TemporaryError

Parameters:

base - Base URL at which the streaming service resides.

auth - Authentication object to use for channel requests.

consumerGroup - Consumer group to subscribe the channel consumer to.

pathPrefix - Path to append to streaming service requests.

consumerPathPrefix - Path to append to consumer-related requests made to the streaming service. Note that if the pathPrefix parameter is set to a non-empty value, the pathPrefix value will be appended to consumer-related requests instead of the consumerPathPrefix value.

retryOnFail - Whether or not the channel will automatically retry a call which failed due to a temporary error.

verifyCertBundle - CA Bundle chain certificates. This string shall be either the certificates themselves or a path to a CA bundle file containing those certificates. The CA bundle is used to validate that the certificate of the authentication server being connected to was signed by a valid authority. If set to an empty string, the server certificate will not be validated.

extraConfigs - Dictionary of key/value pairs containing any custom configuration settings which should be sent to the streaming service when a consumer is created. Examples of key/value pairs are: ("auto.offset.reset", "latest"); ("request.timeout.ms", 30000) and ("session.timeout.ms", 10000).

httpProxySettings - contains http proxy hostname, port, username and password.

Throws:

TemporaryError - if http client request object failed to be created.

Channel

public Channel(java.lang.String base,
               ChannelAuth auth,
               java.lang.String consumerGroup,
               java.lang.String pathPrefix,
               java.lang.String consumerPathPrefix,
               java.lang.String producerPathPrefix,
               boolean retryOnFail,
               java.lang.String verifyCertBundle,
               java.util.Properties extraConfigs,
               HttpProxySettings httpProxySettings)
        throws TemporaryError

Parameters:

base - Base URL at which the streaming service resides.

auth - Authentication object to use for channel requests.

consumerGroup - Consumer group to subscribe the channel consumer to.

pathPrefix - Path to append to streaming service requests.

consumerPathPrefix - Path to append to consumer-related requests made to the streaming service. Note that if the pathPrefix parameter is set to a non-empty value, the pathPrefix value will be appended to consumer-related requests instead of the consumerPathPrefix value.

producerPathPrefix - Path to append to producer-related requests made to the streaming service. Note that if the pathPrefix parameter is set to a non-empty value, the pathPrefix value will be appended to producer-related requests instead of the producerPathPrefix value.

retryOnFail - Whether or not the channel will automatically retry a call which failed due to a temporary error.

verifyCertBundle - CA Bundle chain certificates. This string shall be either the certificates themselves or a path to a CA bundle file containing those certificates. The CA bundle is used to validate that the certificate of the authentication server being connected to was signed by a valid authority. If set to an empty string, the server certificate will not be validated.

extraConfigs - Dictionary of key/value pairs containing any custom configuration settings which should be sent to the streaming service when a consumer is created. Examples of key/value pairs are: ("auto.offset.reset", "latest"); ("request.timeout.ms", 30000) and ("session.timeout.ms", 10000).

httpProxySettings - contains http proxy hostname, port, username and password.

Throws:

TemporaryError - if http client request object failed to be created.

Method Detail

create

public void create()
            throws PermanentError,
                   TemporaryError

Creates a new consumer on the consumer group

Specified by:

create in interface Consumer

Throws:

PermanentError - if no consumer group was specified.

TemporaryError - if the creation attempt fails.

subscribe

public void subscribe(java.util.List<java.lang.String> topics)
               throws ConsumerError,
                      PermanentError,
                      TemporaryError

Subscribes the consumer to a list of topics

Specified by:

subscribe in interface Consumer

Parameters:

topics - Topic list.

Throws:

ConsumerError - if the consumer associated with the channel does not exist on the server.

PermanentError - if no topics were specified.

TemporaryError - if the subscription attempt fails.

subscriptions

public java.util.List<java.lang.String> subscriptions()
                                               throws ConsumerError,
                                                      PermanentError,
                                                      TemporaryError

List the topic names to which the consumer is subscribed.

Specified by:

subscriptions in interface Consumer

Returns:

List of topic names

Throws:

ConsumerError - if the consumer associated with the channel does not exist on the server.

TemporaryError - if the retrieval of subscriptions fails.

PermanentError - if request was malformed.

delete

public void delete()
            throws TemporaryError,
                   PermanentError

Deletes the consumer from the consumer group

Specified by:

delete in interface Consumer

Throws:

TemporaryError - if the delete attempt fails.

PermanentError - if request was malformed.

consume

public ConsumerRecords consume()
                        throws ConsumerError,
                               PermanentError,
                               TemporaryError

Consumes records from all the subscribed topics

Specified by:

consume in interface Consumer

Returns:

ConsumerRecords a list of the consumer record objects from the records returned by the server.

Throws:

ConsumerError - if the consumer associated with the channel does not exist on the server.

PermanentError - if the channel has not been subscribed to any topics.

TemporaryError - if the consume attempt fails.

consume

public ConsumerRecords consume(int timeout)
                        throws ConsumerError,
                               PermanentError,
                               TemporaryError

Consumes records from all the subscribed topics

Specified by:

consume in interface Consumer

Parameters:

timeout - Timeout in milliseconds to wait for records before returning

Returns:

ConsumerRecords a list of the consumer record objects from the records returned by the server.

Throws:

ConsumerError - if the consumer associated with the channel does not exist on the server.

PermanentError - if the channel has not been subscribed to any topics.

TemporaryError - if the consume attempt fails.

commit

public void commit()
            throws ConsumerError,
                   TemporaryError,
                   PermanentError

Commits the record offsets to the channel.

Committed offsets are the latest consumed ones on all consumed topics and partitions.

Specified by:

commit in interface Consumer

Throws:

ConsumerError - if the consumer associated with the channel does not exist on the server.

TemporaryError - if the commit attempt fails.

PermanentError - if request was malformed.

run

public void run(ConsumerRecordProcessor processCallback,
                java.util.List<java.lang.String> topics)
         throws PermanentError,
                TemporaryError

Repeatedly consume records from the subscribed topics.

The supplied ConsumerRecordProcessor.processCallback(ConsumerRecords, String) method is invoked with a list containing each consumer record.

ConsumerRecordProcessor.processCallback(ConsumerRecords, String) return value is currently ignored. It is reserved for future use.

The stop() method can also be called to halt an execution of this method.

Specified by:

run in interface Consumer

Parameters:

processCallback - Callable which is invoked with a list of records which have been consumed.

topics - If set to a non-empty value, the channel will be subscribed to the specified topics. If set to an empty value, the channel will use topics previously subscribed via a call to the subscribe method.

Throws:

PermanentError - if a prior run is already in progress or no consumer group value was specified or callback to deliver records was not specified

TemporaryError - consume or commit attempts failed with errors other than ConsumerError.

run

public void run(ConsumerRecordProcessor processCallback,
                java.util.List<java.lang.String> topics,
                int timeout)
         throws PermanentError,
                TemporaryError

Repeatedly consume records from the subscribed topics.

The supplied ConsumerRecordProcessor.processCallback(ConsumerRecords, String) method is invoked with a list containing each consumer record.

ConsumerRecordProcessor.processCallback(ConsumerRecords, String) return value is currently ignored. It is reserved for future use.

The stop() method can also be called to halt an execution of this method.

Specified by:

run in interface Consumer

Parameters:

processCallback - Callable which is invoked with a list of records which have been consumed.

topics - If set to a non-empty value, the channel will be subscribed to the specified topics. If set to an empty value, the channel will use topics previously subscribed via a call to the subscribe method.

timeout - Timeout in milliseconds to wait for records before returning

Throws:

PermanentError - if a prior run is already in progress or no consumer group value was specified or callback to deliver records was not specified

TemporaryError - consume or commit attempts failed with errors other than ConsumerError.

run

public void run(ConsumerRecordProcessor processCallback,
                java.lang.String topic)
         throws PermanentError,
                TemporaryError

Repeatedly consume records from the subscribed topic.

The supplied ConsumerRecordProcessor.processCallback(ConsumerRecords, String) method is invoked with a list containing each consumer record.

ConsumerRecordProcessor.processCallback(ConsumerRecords, String) return value is currently ignored. It is reserved for future use.

The stop() method can also be called to halt an execution of this method.

Specified by:

run in interface Consumer

Parameters:

processCallback - Callable which is invoked with a list of records which have been consumed.

topic - If set to a non-empty value, the channel will be subscribed to the specified topic. If set to an empty value, the channel will use topics previously subscribed via a call to the subscribe method.

Throws:

PermanentError - if a prior run is already in progress or no consumer group value was specified or callback to deliver records was not specified

TemporaryError - consume or commit attempts failed with errors other than ConsumerError.

run

public void run(ConsumerRecordProcessor processCallback,
                java.lang.String topic,
                int timeout)
         throws PermanentError,
                TemporaryError

Repeatedly consume records from the subscribed topic.

The supplied ConsumerRecordProcessor.processCallback(ConsumerRecords, String) method is invoked with a list containing each consumer record.

ConsumerRecordProcessor.processCallback(ConsumerRecords, String) return value is currently ignored. It is reserved for future use.

The stop() method can also be called to halt an execution of this method.

Specified by:

run in interface Consumer

Parameters:

processCallback - Callable which is invoked with a list of records which have been consumed.

topic - If set to a non-empty value, the channel will be subscribed to the specified topic. If set to an empty value, the channel will use topics previously subscribed via a call to the subscribe method.

timeout - Timeout in milliseconds to wait for records before returning

Throws:

PermanentError - if a prior run is already in progress or no consumer group value was specified or callback to deliver records was not specified

TemporaryError - consume or commit attempts failed with errors other than ConsumerError.

stop

public void stop()
          throws StopError

Stop an active execution of the run(ConsumerRecordProcessor, List) call.

If no run(ConsumerRecordProcessor, List) call is active, this function returns immediately. If a run(ConsumerRecordProcessor, List) call is active, this function blocks until the run has been completed.

Specified by:

stop in interface Consumer

Throws:

StopError - an error occurred while waiting for channel to be stopped

close

public void close()
           throws TemporaryError,
                  StopError,
                  PermanentError

Closes the channel.

It calls destroy() to stop the channel and to release its resources.

This method is added to allow Channel to be used in conjunction with Java try-with-resources statement.

Specified by:

close in interface Consumer

Specified by:

close in interface Producer

Specified by:

close in interface java.lang.AutoCloseable

Throws:

TemporaryError - if a consumer has previously been created for the channel but an attempt to delete the consumer from the channel fails.

StopError - if the attempt to stop the channel fails.

PermanentError

produce

public void produce(ProducerRecords producerRecords)
             throws PermanentError,
                    TemporaryError

Produce records to the channel.

Specified by:

produce in interface Producer

Parameters:

producerRecords - a ProducerRecords object containing the records to be posted to the channel.

Throws:

PermanentError - if produce request was malformed or produce RESTful service was not found.

TemporaryError - if produce request was temporarily not authorized or there was an internal RESTful error while serving the request.

produce

public void produce(java.lang.String jsonProducerRecords)
             throws PermanentError,
                    TemporaryError

Produce records to the channel.

Specified by:

produce in interface Producer

Parameters:

jsonProducerRecords - a String object containing the records to be posted to the channel in JSON string format.

Throws:

PermanentError - if produce request was malformed or produce RESTful service was not found.

TemporaryError - if produce request was temporarily not authorized or there was an internal RESTful error while serving the request.