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 results of the operation
    DequeueResult dequeue();

    /// 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)
    ///  - The number of Milliseconds specified has elapsed (DequeueResult will have a status of TIMEOUT)
    ///  - A network error occurs (StreamBaseException will be thrown)
    ///
    /// A timeout_ms of zero will block indefinitely, or until a tuple arrives.
    /// Note that the actual dequeue timeout may vary based on normal thread scheduling,
    /// plus network interactions with the server.
    ///
    /// @return a DequeueResult containing the results of the operation
    DequeueResult dequeue(int timeout_ms);

    ///
    /// Enqueue tuples to a stream.  The stream must not be tied to
    /// the output of any operator in the application. This method can block
    /// depending on network, or StreamBase server, congestion.
    ///
    /// Note: enqueue() will modify the tuple as it is enqueued.  If you
    /// do not want the tuple modified you must make a copy of it before
    /// calling enqueue.
    ///
    /// @deprecated Use StreamProperties based enqueue
    /// @param stream_name the name of the stream to which to enqueue
    /// @param tuple a tuple to enqueue
    void enqueue(const std::string &stream_name, Tuple& tuple) ;

    ///
    /// Enqueue tuples to a stream.  The stream must not be tied to
    /// the output of any operator in the application. This method can block
    /// depending on network, or StreamBase server, congestion.
    ///
    /// Note: enqueue() will modify the tuple as it is enqueued.  If you
    /// do not want the tuple modified you must make a copy of it before
    /// calling enqueue.
    ///
    /// @param props the StreamProperties to enqueue the tuple to
    /// @param tuple a tuple to enqueue
    void enqueue(const StreamProperties &props, Tuple& tuple) ;

    /// Enqueue tuples to a stream.  The stream must not be tied to
    /// the output of any operator in the application. This method can block
    /// depending on network, or StreamBase server, congestion.
    ///
    /// Note: enqueue() will modify the tuples as they are enqueued.  If you
    /// do not want the tuples modified you must make copies  before
    /// calling enqueue.
    /// 
    /// @deprecated Use StreamProperties based enqueue
    /// @param stream_name the name of the stream to which to enqueue
    /// @param tuples a list of tuples to enqueue
    void enqueue(const std::string &stream_name, TupleList& tuples) ;

    /// Enqueue tuples to a stream.  The stream must not be tied to
    /// the output of any operator in the application. This method can block
    /// depending on network, or StreamBase server, congestion.
    ///
    /// Note: enqueue() will modify the tuples as they are enqueued.  If you
    /// do not want the tuples modified you must make copies  before
    /// calling enqueue.
    ///
    /// @param props the StreamProperties to enqueue the tuples to
    /// @param tuples a list of tuples to enqueue
    void enqueue(const StreamProperties &props, TupleList& tuples) ;
#endif

    /// Returns a schema with a particular hash.
    Schema getSchemaByHash(const std::string &hash);

    /// Returns a schema by name. This will only succeed for named schemas;
    /// anonymous schemas assigned to a port should instead be looked up
    /// via getStreamProperties().getSchema().
    Schema getSchemaByName(const std::string &name);

    /// status of java operators and adapters
    void operatorStatus(const std::string &containerName, std::vector<std::string> &aStatus);

    /// status the streambase daemons
    void status(std::vector<std::string> &aStatus, bool verbose = false);

    /// Return the URI we're talking to.
    const StreamBaseURI& getURI() const;

    const std::vector<StreamBaseURI>& getURIs() const;

    /// Terminate the client.  May be invoked from a different thread.
    /// StreamBaseClient memory, network, and thread resources are not released
    /// until close() is called. 
    ///
    /// Returns immediately.
    void close();

    /// Return true if the client connection is closed.
    /// The client connection can be closed by calling the close() method,
    /// or by a server shutdown, or a network error.
    bool isClosed();

    
    /// 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.
    /// @return boolean if we can dequeue without blocking
    bool canDequeue();

    /// Enable heartbeating for this client.  Generally this is only for enqueue-only clients.
    void enableHeartbeating();


    /// Returns true if this TupleConnections has any live connections to a
    /// server. Note that a return of "false" doesn't necessarily indicate an
    /// error since the TupleConnection's state (e.g., lack of subscriptions)
    /// might mean no connections are needed.
    /// @return boolean if we are connected
    bool isConnected();

    /// Turn on buffering. The WakeAndFlushBuffer thread is only started if flush_interval > 0.
    void enableBuffering(int buffer_size, int flush_interval = DEFAULT_FLUSH_INTERVAL);

#ifndef DOXYGEN_SKIP
    /// Enable connection less enqueue.  This enables a mode where tuples are enqueued
    /// over an XMLRPC request rather than a binary connection.  It is useful for low
    /// volume enqueue of tuples.  It may reduce the load on the server as the connection
    /// is transient.  Setting has no effect upon dequeue.
    /// @param enable enable connection less enqueue of tuples.  Default is off.
    /// excluded until server performance improves
    /// <em>Note: this method is not public API, and is for internal StreamBase use only</em>
    void enableConnectionlessEnqueue(bool enable);
#endif

#ifndef DOXYGEN_SKIP
    /// Return state of connection less enqueue setting
    /// excluded until server performance improves
    /// <em>Note: this method is not public API, and is for internal StreamBase use only</em>
    bool getIsConnectionlessEnqueue();
#endif

    /// Flush any pending enqueue buffers.
    /// This operation has no effect if buffering is not enabled.
    ///
    /// @throws StreamBaseException if there is an IO error while flushing the buffer
    void flushAllBuffers();

    /// Flush any pending enqueue buffer for the StreamProperties provided.
    /// This operation has no effect if buffering is not enabled or 
    /// there is no buffer to flush for the given stream.
    /// <br/>
    /// <b>Note:</b> Note that this will cause inter-stream ordering to be interrupted.
    /// <br/>
    /// @param props the stream whose enqueue buffers to flush, if not empty
    /// @throws StreamBaseException if there is an IO error while flushing the buffer
    /// @deprecated use flushAllBuffers() to preserve inter-stream ordering 
    void flushBuffer(const StreamProperties &props);

    /// Flush any pending enqueue buffer for the stream name provided.
    /// This operation has no effect if buffering is not enabled or 
    /// there is no buffer to flush for the given stream.
    /// <br/>
    /// <b>Note:</b> Note that this will cause inter-stream ordering to be interrupted.
    /// <br/>
    /// @param stream_name the stream whose enqueue buffers to flush, if not empty
    /// @throws StreamBaseException if there is an IO error while flushing the buffer
    /// @deprecated stream_name use StreamProperties based flushBuffer
    /// @deprecated use flushAllBuffers() to preserve inter-stream ordering 
    void flushBuffer(const std::string &stream_name);

    /// Return whether buffering is turned on or off
    bool getIsBuffering() const;

    /// Return buffer size (in tuples)
    int  getBufferSize() const;

    /// Return the Connection ID for this Client Connection.  Only Valid once
    /// an enqueue/dequeue has been attempted. 
    /// @returns connection ID in binary format
    std::string getConnectionID() const;

    /// Get all the dynamic variables in the given module, and return them
    /// as a Tuple where the field names are the names of the dynamic variables, 
    /// and the field values are the current values of the dynamic variables.
    /// 
    /// @throws StreamBaseException on network or other errors, or if there are
    /// no dynamic variables in the named module
    Tuple getDynamicVariables(const std::string& modulePath);

    /// Set the dynamic variable at the given path to the given value, expressed 
    /// as a string in CSV style.
    /// @throws StreamBaseException if the dynamic variable does not exist, or the
    /// value is not appropriate for setting the dynamic variable
    void setDynamicVariable(const std::string& dynamicVariablePath, const std::string& value);

    /// Return the Connection ID for this Client Connection.  Only Valid once
    /// an enqueue/dequeue has been attempted. 
    /// @returns connection ID in Hexadecimal format
    std::string getConnectionIdAsHexString() const;

    /// Read rows out of the table at this path. Limit to the
    /// specified number of rows returned, or -1 for no limit. Filter
    /// rows according to predicate, or return all rows if predicate
    /// is empty.
    /// @returns a list of matching rows.
    TupleList readTable(const std::string &tablePath, int limit, const std::string& predicate="");

#ifndef DOXYGEN_SKIP
    /// Execute an internal command.  Internal commands are subject to change and
    /// for StreamBase internal use only.
    void internalCommand(const std::vector<std::string>& arguments, std::vector<std::string> &result);

    bool doByteSwap();
#endif

    /// Sets how many milliseconds a dequeueing client will tolerate not receiving a client heart 
    /// beat from the StreamBase server that it is connected to. The default value is
    /// 120 seconds (120000). By default, StreamBase servers emit "client heart beats"
    /// every 10 seconds so StreamBase applications have no requirement to send data regularly.
    ///
    /// @param timeoutMS The number of milliseconds that is allowed to pass without
    /// receiving a message from the StreamBase server client heart beat stream.
    int setQuiescentLimit(int timeoutMS);

    /// Returns the client version as a string
    std::string getVersion() {
      return std::string(StreamBaseVersion::INFO_LINE);
    }

    ///
    /// Get the settings for the client
    ///
    sb::ClientSettings &getSettings() {
        return _clientSettings;
    }

    /// Returns the number of tuples this client has actually enqueued to the server.
    /// Note that this number will not include any tuples that have been buffered but
    /// have not actually been enqueued to the server.
    long long getTupleEnqueueCount() const;

    /// Returns the number of tuples this client has dequeued from the server. This number
    /// does not include tuples that may be dequeued as part of system services such 
    /// as heartbeating.  This number does however include tuples that have been dequeued
    /// but have been intercepted by the Interceptor.
    long long getTupleDequeueCount() const;

private:
    typedef std::map<std::string, StreamProperties> StreamMap;

    std::shared_ptr<sb_internal::StreamBaseClientImpl> _impl;

    // No copying.
    StreamBaseClient(const StreamBaseClient&);

    // No copying.
    StreamBaseClient& operator = (const StreamBaseClient&);

    void init();

    // keep track of logical and associated real streams
    StreamMap _streamMap;

    sb::ClientSettings _clientSettings;
};


SB_NAMESPACE_END;
#endif