StreamBaseClient.hpp

// Copyright (c) 2004-2023 TIBCO Software Inc.  All rights reserved.
 
#ifndef STREAMBASE_CLIENT_H
#define STREAMBASE_CLIENT_H
 
#include "StreamBase.hpp"
#include "StreamBaseVersion.hpp"
#include "Exceptions.hpp"
#include "Schema.hpp"
#include "Field.hpp"
#include "StreamProperties.hpp"
#include "StreamBaseEntityType.hpp"
#include "StreamBaseURI.hpp"
#include "TupleList.hpp"
#include "DequeueResult.hpp"
#include "ClientSettings.hpp"
 
SB_INTERNAL_FWD(StreamBaseClientImpl)
 
SB_NAMESPACE_BEGIN;
 
/// The StreamBase client API.  Connects to an StreamBase node over the network,
/// via XML/RPC and the tuple wire protocol. With the exception of the close() method,
/// the StreamBaseClient object is not thread safe, so access to a single object
/// needs to be synchronized between threads. If multiple threads subscribe to streams,
/// it is recommended that they use separate instances of this class.
///
class StreamBaseClient {
  public:
    static const unsigned int DEFAULT_FLUSH_INTERVAL = 250;
 
#ifndef DOXYGEN_SKIP
    /// This tells the rest of the API that tuples want to remain in serialized form,
    /// ostensibly for consumption by the .NET API.
    static bool useRawTuples;
#endif
 
    /// Creates an StreamBase client and establishes a connection to a remote
    /// server. Optional parameter buffer_size specifies the number of tuples
    /// to buffer before enqueueing. Optional parameter flush_interval specifies
    /// the interval in milliseconds between wakeups of the wakeAndFlushBuffer.
    /// The wakeAndFlushBuffer thread is only started if buffer_size > 0.
    StreamBaseClient(const StreamBaseURI& uri = StreamBaseURI::fromEnvironment(),
                     int buffer_size = 0,
                     int flush_interval=DEFAULT_FLUSH_INTERVAL);
 
    StreamBaseClient(const std::string& uris,
                     int buffer_size = 0,
                     int flush_interval=DEFAULT_FLUSH_INTERVAL);
 
    StreamBaseClient(const std::vector<StreamBaseURI>& uris,
                     int buffer_size = 0,
                     int flush_interval=DEFAULT_FLUSH_INTERVAL);
 
    StreamBaseClient(const std::vector<StreamBaseURI>& uris,
                     sb::ClientSettings &settings,
                     int buffer_size = 0,
                     int flush_interval=DEFAULT_FLUSH_INTERVAL);
 
    /// Destroys a session.
    ~StreamBaseClient();
 
    /// Flags for the listEntities call
    /// This must be consistent with what is found on the Java side of the 
    ///  house in StreamBaseClient.java
    enum ListEntitiesFlags {
        /// set FULLY_QUALIFIED_NAMES if you want to 
        /// include container names for all entities.  If not set then the returned 
        /// names will contain the module name (if any) and the name of the entity.
        FULLY_QUALIFIED_NAMES = 1,
        /// set INCLUDE_MODULES if you want to include all modules in the output 
        INCLUDE_MODULES = 2,
        /// set ALL_CONTAINERS if you want to include all 
        /// user containers in the output.  This does not include the "system" 
        /// container.
        ALL_CONTAINERS = 4 | FULLY_QUALIFIED_NAMES
    };
 
    /// Lists all entities of a particular type.  (One can then use
    /// "describe" to obtain a description of an entity.)  Names is
    /// cleared first. Use container.entity-type to resolve across containers.
    /// use the given flags to list the entities
    void listEntities(const std::string &type, int flags, std::vector<std::string>& names);
 
    /// Lists all entities of a particular type.  (One can then use
    /// "describe" to obtain a description of an entity.)  names is
    /// cleared first.
    /// use the given flags to list the entities
    void listEntities(StreamBaseEntityType::Type type, int flags, std::vector<std::string>& names);
 
    /// Lists all entities of a particular type.  (One can then use
    /// "describe" to obtain a description of an entity.)  names is
    /// cleared first.
    void listEntities(StreamBaseEntityType::Type type, std::vector<std::string>& names);
 
    /// Returns an XML description of an entity, or an empty string if
    /// the method does not exist.
    std::string describe(const std::string &entity);
 
#ifndef DOXYGEN_SKIP
    /// Returns true if the license for the specified feature is available and false otherwise.
    bool checkLicense(const std::string &featureName);
#endif
 
    /// Returns true if this stream has been subscribed to.
    /// @param entity    the path of the stream
    bool isStreamSubscribed(const std::string &entity);
 
    /// Returns a description of a stream, throwing an exception if the
    /// stream does not exist.
    /// @param entity    the path of the stream
    /// @param strategy  how to handle capture fields on the stream
    StreamProperties getStreamProperties(const std::string &entity,
            StreamProperties::CaptureTransformStrategy strategy=StreamProperties::FLATTEN);
 
    /// Returns a description of a stream, throwing an exception if the
    /// stream does not exist.
    StreamProperties getStreamPropertiesByHash(const std::string &hex);
 
    /// Returns the schema of a stream, throwing an exception if the
    /// stream does not exist.
    Schema getSchemaForStream(const std::string &entity,
            StreamProperties::CaptureTransformStrategy strategy=StreamProperties::FLATTEN);
 
    /// Typechecks a potential modification to the application, outputting
    /// properties of all streams in the application.  streams is cleared
    /// first.
    /// @param sbapp the application
    /// @param streams the schema defs of the streams
    /// @param full do a full typecheck
    void typecheck(const std::string &sbapp,
                   std::map<std::string, Schema>& streams,
                   bool full = false);
 
    /// Subscribes to a stream.
    ///
    /// Uses the default capture transform strategy of flatten. To specify
    /// capture strategy, use getStreamProperties(stream_name,
    /// StreamProperties::NEST) and subscribe using the resultant
    /// StreamProperties.
    ///
    /// @return the StreamProperties for the stream
    StreamProperties subscribe(const std::string &stream_name);
    /// Subscribes to a stream.
    /// @return the StreamProperties for the stream
    StreamProperties subscribe(const StreamProperties &props);
 
    /// Subscribes to a stream with a predicate to apply to output tuples.  
    /// The stream name of dequeued tuples is logical_stream.
    /// When unsubscribing, use logical_stream.
    ///
    /// Uses the default capture transform strategy of flatten. To specify
    /// capture strategy, use getStreamProperties(stream_name,
    /// StreamProperties::NEST) and subscribe using the resultant
    /// StreamProperties.
    ///
    /// @param stream_name the stream to subscribe to, error if empty
    /// @param logical_stream the name of the logical stream to associate with this predicate (if empty, defaults to streamname)
    /// @param predicate a predicate to apply to subset the stream, error if empty
    /// @return the StreamProperties for the stream
    StreamProperties subscribe(const std::string &stream_name, const std::string &logical_stream, const std::string &predicate);
 
    /// Subscribes to a stream with a predicate to apply to output tuples.  
    /// The stream name of dequeued tuples is logical_stream.
    /// When unsubscribing, use logical_stream.
    /// @param props the stream to subscribe to, error if empty
    /// @param logical_stream the name of the logical stream to associate with this predicate (if empty, defaults to streamname)
    /// @param predicate a predicate to apply to subset the stream, error if empty
    /// @return the StreamProperties for the stream
    StreamProperties subscribe(const StreamProperties &props, const std::string &logical_stream, const std::string &predicate);
    
    /// Resubscribes to an already subscribed stream
    ///
    /// Uses the default capture transform strategy of flatten. To specify
    /// capture strategy, use getStreamProperties(stream_name,
    /// StreamProperties::NEST) and subscribe using the resultant
    /// StreamProperties.
    ///
    /// @return the StreamProperties for the stream
    StreamProperties resubscribe(const std::string &stream_name, const std::string &logical_stream, const std::string &predicate);
 
    /// Resubscribes to an already subscribed stream
    /// @return the StreamProperties for the stream
    StreamProperties resubscribe(const StreamProperties &props, const std::string &logical_stream, const std::string &predicate);
 
    /// UnSubscribes to a stream.  Note: Any tuples that are in-flight during 
    /// an unsubscribe request will be dequeued until the stream is fully drained.
    void unsubscribe(const std::string &stream_name);
 
    /// UnSubscribes to a stream.  Note: Any tuples that are in-flight during 
    /// an unsubscribe request will be dequeued until the stream is fully drained.
    void unsubscribe(const StreamProperties &props);
 
#ifndef SWIG
    ///
    /// 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().
    ///
    /// NOTE: Once the DequeueResult.Interceptor is passed to StreamBaseClient,
    /// StreamBaseClient will manage the lifecycle of the object.  StreamBaseClient
    /// will delete the object when the interceptor is reset or when the StreamBaseClient
    /// is deleted.
    /// Do not delete this passed object!!
    void setDequeueResultInterceptor(DequeueResult::Interceptor *dri);
 
    ///
    /// Get the current dequeue results interceptor, or null if there is no
    /// current processor.
    ///
    DequeueResult::Interceptor *getDequeueResultsInterceptor();
    
    /// Dequeue tuples from any subscribed stream.  This method
    /// blocks until
    ///
    ///  - at least one tuple is available on any subscribed stream, or
    ///  - the node is shut down.
    ///
    /// @deprecated Use DequeueResult based dequeue
    /// @param streamProperties (output) set to the StreamProperties of the stream from
    ///     which tuples have been dequeued
    /// @return a list of the tuples dequeued; empty if the node has been
    ///     shut down
    TupleList dequeue(StreamProperties &streamProperties);
 
    /// Dequeue tuples from any subscribed stream.  This method
    /// blocks until
    ///
    ///  - at least one tuple is available on any subscribed stream, or
    ///  - the node is shut down.
    ///
    /// @deprecated Use DequeueResult based dequeue
    /// @param stream_name (output) set to the name of the stream from
    ///     which tuples have been dequeued.  Note this does not contain the container name
    /// @return a list of the tuples dequeued; empty if the node has been
    ///     shut down
    TupleList dequeue(std::string &stream_name);
 
    /// Dequeue tuples from any subscribed stream.  This method
    /// blocks until
    ///
    ///  - At least one tuple is available on any subscribed stream (DequeueResult will have a status of GOOD)
    ///  - The StreamBase server the client is connected to is shut down (DequeueResult will have a status of CLOSED)
    ///  - A network error occurs (StreamBaseException will be thrown)
    ///
    /// @return a DequeueResult containing the r