com.streambase.sb.client

Class DeMUXStreamBaseClient

java.lang.Object

com.streambase.sb.client.BaseClient

com.streambase.sb.client.StreamBaseClient

com.streambase.sb.client.DeMUXStreamBaseClient

All Implemented Interfaces:

AutoCloseable

public class DeMUXStreamBaseClient
extends StreamBaseClient

This is a StreamBaseClient that is intended for use by several independent queries. Rather than polling for data, when you subscribe you register a DequeueListener, which will be handed only the results for YOUR subscribe call

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
static interface	DeMUXStreamBaseClient.DequeueListener Implement this interface to get calls back with data when using a DeMUXStreamBaseClient

Nested classes/interfaces inherited from class com.streambase.sb.client.StreamBaseClient

StreamBaseClient.ListEntityFlags, StreamBaseClient.SerializedTupleBuffer, StreamBaseClient.TupleBuffer

Field Summary

Fields

Modifier and Type	Field and Description
protected static long	DEQUEUE_WAIT
(package private) Map<String,DeMUXStreamBaseClient.DequeueListener>	listenerMap

Constructor Summary

Constructors

Constructor and Description
DeMUXStreamBaseClient(List<StreamBaseURI> uris, ClientSettings settings)
DeMUXStreamBaseClient(StreamBaseURI streamBaseURI)
DeMUXStreamBaseClient(String uri)

Method Summary

Modifier and Type	Method and Description
boolean	canDequeue() Return true if we can call dequeue without blocking.
DequeueResult	dequeue() Dequeue a batch of tuples from a subscribed stream.
DequeueResult	dequeue(long timeoutMS) Dequeue a batch of tuples from a subscribed stream.
int	getDequeueBufferSize() Return the number of tuples that are in the dequeue buffer.
protected static String	getFullLogicalName(StreamProperties streamProperties)
String	getUniqueSuffix() This manages a threadsafe counter which will give a different value each time it is called.
boolean	isSubscribed(StreamProperties stream) Return status if we are subscribed to the given stream
StreamProperties	resubscribe(StreamProperties props, String logicalstream, String predicate) Resubscribe to a stream with a predicate.
StreamProperties	resubscribe(String streamname, String logicalstream, String predicate) Resubscribe to a stream with a predicate.
DequeueResultsQueue	setDequeueQueue(DequeueResultsQueue dequeuer) Set the queue that is used to dequeue tuples from remote servers.
DequeueResult.Interceptor	setDequeueResultInterceptor(DequeueResult.Interceptor dri) Set the dequeue results interceptor for this client connection.
StreamProperties	subscribe(StreamProperties props) Subscribe to a stream
StreamProperties	subscribe(StreamProperties props, String logicalstream, String predicate) Subscribe to a stream with a predicate.
StreamProperties	subscribe(String streamname) Subscribe to a stream
StreamProperties	subscribe(String streamName, CaptureTransformStrategy strategy) Subscribe to a stream
StreamProperties	subscribe(String streamName, CaptureTransformStrategy strategy, String logicalstream, String predicate) Subscribe to a stream with a predicate.
StreamProperties	subscribe(String streamname, String logicalstream, String predicate) Subscribe to a stream with a predicate.
StreamProperties	subscribe(String streamname, String logicalstream, String predicate, DeMUXStreamBaseClient.DequeueListener listener) This is the only subscribe operation that is supported by the DeMUXStreamBaseClient
void	unsubscribe(StreamProperties logicalProps) Unsubscribe from the given stream name.
void	unsubscribe(String logicalStreamName) Unsubscribe from the given stream name.

Methods inherited from class com.streambase.sb.client.StreamBaseClient

addConnectionStatusCallback, assureNameStartsWithContainer, checkValidStreamname, close, close, connectionIDObject, describe, enableBuffering, enableBuffering, enqueue, enqueue, enqueue, enqueue, flushAllBuffers, flushBuffer, flushBuffer, getAllStreamProperties, getAllStreamProperties, getConnectionError, getConnectionID, getConnectionStatus, getDequeuer, getDequeueResultsInterceptor, getDynamicVariable, getDynamicVariables, getEnqueueBufferSize, getSchemaByHash, getSchemaByName, getSchemaForStream, getStreamProperties, getStreamProperties, getStreamPropertiesByHash, getStreamPropertiesByHash, getSubscribedStreamNames, getSubscribedStreams, getTupleDequeueCount, getTupleEnqueueCount, getVersion, hasSchema, hasStream, hasStreamProperties, incrementTuplesDequeued, incrementTuplesEnqueued, isClosed, listEntities, listEntities, listEntities, listEntities, listEntities, listEntities, readTable, readTable, removeConnectionStatusCallback, setConnectionID, setConnectionStatus, setDynamicVariable, setQuiescentLimit, typecheck, typecheck, unsubscribeInternal

Methods inherited from class com.streambase.sb.client.BaseClient

getSettings, getURI, getURIs, haModeOn, operatorStatus, status, status

Methods inherited from class java.lang.Object

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

Field Detail

listenerMap

Map<String,DeMUXStreamBaseClient.DequeueListener> listenerMap

DEQUEUE_WAIT

protected static final long DEQUEUE_WAIT

See Also:

Constant Field Values

Constructor Detail

DeMUXStreamBaseClient

public DeMUXStreamBaseClient(List<StreamBaseURI> uris,
                             ClientSettings settings)
                      throws URIException,
                             StreamBaseException

Throws:

StreamBaseException

URIException

DeMUXStreamBaseClient

public DeMUXStreamBaseClient(String uri)
                      throws URIException,
                             StreamBaseException

Throws:

URIException

StreamBaseException

DeMUXStreamBaseClient

public DeMUXStreamBaseClient(StreamBaseURI streamBaseURI)
                      throws URIException,
                             StreamBaseException

Throws:

URIException

StreamBaseException

Method Detail

subscribe

public StreamProperties subscribe(String streamname,
                                  String logicalstream,
                                  String predicate,
                                  DeMUXStreamBaseClient.DequeueListener listener)
                           throws StreamBaseException

This is the only subscribe operation that is supported by the DeMUXStreamBaseClient

Parameters:

streamname - Qualified or partial streamname. If partial, will be relative to the connection base (usually default.)

logicalstream - Logical name for this stream. If you specify a predicate, this is required and must be different from streamname Also, only one listener per logicalStream name is allowed for a given DeMUXStreamBaseClient

predicate - Your query

listener - Implement this interface to be called when your data arrives (and other events)

Returns:

StreamProperties for the stream being listened to.

Throws:

StreamBaseException

getFullLogicalName

protected static String getFullLogicalName(StreamProperties streamProperties)

getUniqueSuffix

public String getUniqueSuffix()

This manages a threadsafe counter which will give a different value each time it is called.

Returns:

a string with underscore and a number. Each call to it will be different.

canDequeue

public boolean canDequeue()

Description copied from class: StreamBaseClient

Return true if we can call dequeue without blocking. This means that there is something to dequeue from the server. This dequeued item could be Tuples, a null (server shutdown) or an exception.

Overrides:

canDequeue in class StreamBaseClient

Returns:

boolean if we can dequeue without blocking

dequeue

public DequeueResult dequeue()
                      throws StreamBaseException

Description copied from class: StreamBaseClient

Dequeue a batch of tuples from a subscribed stream. This method will block. Will return null if the connection is closed.

Overrides:

dequeue in class StreamBaseClient

Returns:

a DequeueResult (or null) from any subscribed stream (only one stream per DequeueResult)

Throws:

StreamBaseException - thrown on network or other errors

dequeue

public DequeueResult dequeue(long timeoutMS)
                      throws StreamBaseException

Description copied from class: StreamBaseClient

Dequeue a batch of tuples from a subscribed stream. This method will block until there are tuples available or timeoutMS milliseconds have past. It will return null if the connection is closed. It will return an empty DequeueResult if timeoutMS milliseconds have past and no tuples have returned. A timeout_ms of zero will block indefinitely, or until a tuple arrives.

Overrides:

dequeue in class StreamBaseClient

Parameters:

timeoutMS - timeout in milliseconds

Returns:

a DequeueResult from any subscribed stream (only one stream per DequeueResult)

Throws:

StreamBaseException - thrown on network or other errors

setDequeueQueue

public DequeueResultsQueue setDequeueQueue(DequeueResultsQueue dequeuer)

Description copied from class: StreamBaseClient

Set the queue that is used to dequeue tuples from remote servers. This is advanced functionality that could cause fatal errors if implemented incorrectly. Note that any DequeueResults on the old DequeueResultsQueue will be lost when the DequeueResultsQueue is changed.

Overrides:

setDequeueQueue in class StreamBaseClient

Parameters:

dequeuer - new DequeueResultsQueue

Returns:

the old DequeueQueue

setDequeueResultInterceptor

public DequeueResult.Interceptor setDequeueResultInterceptor(DequeueResult.Interceptor dri)

Description copied from class: StreamBaseClient

Set the dequeue results interceptor for this client connection. This results interceptor replaces any existing results processor. To disable pre-processing of results, set the processor to null. This method cannot be safely called while another thread is calling dequeue().

Overrides:

setDequeueResultInterceptor in class StreamBaseClient

Parameters:

dri - dequeue results interceptor for this client connection

Returns:

the old dequeue results interceptor

subscribe

public StreamProperties subscribe(String streamname)
                           throws StreamBaseException

Description copied from class: StreamBaseClient

Subscribe to a stream

Overrides:

subscribe in class StreamBaseClient

Parameters:

streamname - the stream to subscribe to

Returns:

the stream properties

Throws:

StreamBaseException - thrown on error

subscribe

public StreamProperties subscribe(String streamName,
                                  CaptureTransformStrategy strategy)
                           throws StreamBaseException

Description copied from class: StreamBaseClient

Subscribe to a stream

Overrides:

subscribe in class StreamBaseClient

Parameters:

streamName - the name of a stream, that may be fully qualified (for example, mycontainer.MyInputSchema). When unqualified, the container from the URI the connection was established with is used, otherwise StreamBaseURI.DEFAULT_CONTAINER is used.

strategy - the capture transform strategy

Returns:

the stream properties

Throws:

StreamBaseException - thrown on error

subscribe

public StreamProperties subscribe(StreamProperties props)
                           throws StreamBaseException

Description copied from class: StreamBaseClient

Subscribe to a stream

Overrides:

subscribe in class StreamBaseClient

Parameters:

props - the stream to subscribe to

Returns:

the stream properties

Throws:

StreamBaseException - thrown on error

subscribe

public StreamProperties subscribe(String streamname,
                                  String logicalstream,
                                  String predicate)
                           throws StreamBaseException

Description copied from class: StreamBaseClient

Subscribe to a stream with a predicate. The stream name of dequeued tuples is logicalstream. When unsubscribing, use logicalstream.

Overrides:

subscribe in class StreamBaseClient

Parameters:

streamname - the stream to subscribe to

logicalstream - the name of the logical stream to associate with this predicate (if empty or null, defaults to streamname)

predicate - a predicate to apply to subset the stream

Returns:

the stream properties

Throws:

StreamBaseException - thrown on error, including empty or null streamname or predicate

subscribe

public StreamProperties subscribe(String streamName,
                                  CaptureTransformStrategy strategy,
                                  String logicalstream,
                                  String predicate)
                           throws StreamBaseException

Description copied from class: StreamBaseClient

Subscribe to a stream with a predicate. The stream name of dequeued tuples is logicalstream. When unsubscribing, use logicalstream.

Overrides:

subscribe in class StreamBaseClient

Parameters:

streamName - the name of a stream, that may be fully qualified (for example, mycontainer.MyInputSchema). When unqualified, the container from the URI the connection was established with is used, otherwise StreamBaseURI.DEFAULT_CONTAINER is used.

strategy - the CaptureTransformStrategy to use if there are capture fields in the stream

logicalstream - the name of the logical stream to associate with this predicate (if empty or null, defaults to streamname)

predicate - a predicate to apply to subset the stream

Returns:

Stream properties

Throws:

StreamBaseException - thrown on error, including empty or null streamname or predicate

subscribe

public StreamProperties subscribe(StreamProperties props,
                                  String logicalstream,
                                  String predicate)
                           throws StreamBaseException

Description copied from class: StreamBaseClient

Subscribe to a stream with a predicate. The stream name of dequeued tuples is logicalstream. When unsubscribing, use logicalstream.

Overrides:

subscribe in class StreamBaseClient

Parameters:

props - the stream to subscribe to

logicalstream - the name of the logical stream to associate with this predicate (if empty or null, defaults to streamname)

predicate - a predicate to apply to subset the stream

Returns:

stream properties

Throws:

StreamBaseException - thrown on error, including empty or null streamname or predicate

resubscribe

public StreamProperties resubscribe(String streamname,
                                    String logicalstream,
                                    String predicate)
                             throws StreamBaseException

Description copied from class: StreamBaseClient

Resubscribe to a stream with a predicate. Allows the client to change the filtered subscribe predicate atomically so that currently buffered data is not lost (as would happen if you unsubscribed, then subscribed again with a different predicate).

Overrides:

resubscribe in class StreamBaseClient

Parameters:

streamname - the stream to subscribe to

logicalstream - the name of the logical stream to associate with this predicate

predicate - a predicate to apply to subset the stream

Returns:

stream properties

Throws:

StreamBaseException - thrown on error

resubscribe

public StreamProperties resubscribe(StreamProperties props,
                                    String logicalstream,
                                    String predicate)
                             throws StreamBaseException

Description copied from class: StreamBaseClient

Resubscribe to a stream with a predicate. Allows the client to change the filtered subscribe predicate atomically so that currently buffered data is not lost (as would happen if you unsubscribed, then subscribed again with a different predicate).

Overrides:

resubscribe in class StreamBaseClient

Parameters:

props - the stream to subscribe to

logicalstream - the name of the logical stream to associate with this predicate

predicate - a predicate to apply to subset the stream

Returns:

stream properties

Throws:

StreamBaseException - thrown on error

unsubscribe

public void unsubscribe(String logicalStreamName)
                 throws StreamBaseException

Description copied from class: StreamBaseClient

Unsubscribe from the given stream name. Note: Any tuples that are in-flight during an unsubscribe request will be dequeued until the stream is fully drained.

Overrides:

unsubscribe in class StreamBaseClient

Parameters:

logicalStreamName - the name of the stream to unsubscribe from, which is logical streamname on filtered subscribe

Throws:

StreamBaseException - thrown on network or other errors

unsubscribe

public void unsubscribe(StreamProperties logicalProps)
                 throws StreamBaseException

Description copied from class: StreamBaseClient

Unsubscribe from the given stream name. Note: Any tuples that are in-flight during an unsubscribe request will be dequeued until the stream is fully drained.

Overrides:

unsubscribe in class StreamBaseClient

Parameters:

logicalProps - the StreamProperties for this logical subscription

Throws:

StreamBaseException - thrown on network or other errors

isSubscribed

public boolean isSubscribed(StreamProperties stream)

Description copied from class: StreamBaseClient

Return status if we are subscribed to the given stream

Overrides:

isSubscribed in class StreamBaseClient

Parameters:

stream - the stream we want to check

Returns:

status if we are subscribed to the given stream

getDequeueBufferSize

public int getDequeueBufferSize()

Description copied from class: StreamBaseClient

Return the number of tuples that are in the dequeue buffer.

Overrides:

getDequeueBufferSize in class StreamBaseClient

Returns:

number of tuples in the the dequeue buffer