Package 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

  • Field Details

  • Constructor Details

  • Method Details

    • 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 stream name. 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 - subscription error

    • getFullLogicalName

      protected static String getFullLogicalName(StreamProperties streamProperties)
      Get full logical name
      Parameters:
      streamProperties - stream properties
      Returns:
      logical name

    • 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 com.streambase.sb.client.DequeueResultsQueue setDequeueQueue(com.streambase.sb.client.DequeueResultsQueue dequeuer)

    • 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()