StreamingClient.h

Go to the documentation of this file.

/*
 * Copyright (c) Meta Platforms, Inc. and affiliates.
 *
 * This source code is licensed under the MIT license found in the
 * LICENSE file in the root directory of this source tree.
 */
 
#ifndef FACEBOOK_NETWORK_STREAMING_CLIENT_H
#define FACEBOOK_NETWORK_STREAMING_CLIENT_H
 
#include "ocean/network/Network.h"
#include "ocean/network/Address4.h"
#include "ocean/network/Port.h"
#include "ocean/network/Streaming.h"
#include "ocean/network/TCPClient.h"
#include "ocean/network/PackagedUDPServer.h"
 
#include "ocean/base/Callback.h"
#include "ocean/base/Lock.h"
 
#include <queue>
 
namespace Ocean
{
 
namespace Network
{
 
/**
 * This class implements a streaming client.<br>
 * The client uses a TCP connection for configuration tasks and a UDP connection for the data transfer.
 * @ingroup network
 */
class OCEAN_NETWORK_EXPORT StreamingClient : public Streaming
{
    public:
 
        /**
         * Definition of a callback function for streaming server requests.
         */
        using RequestCallback = Callback<bool, const State, const std::string&>;
 
        /**
         * Definition of a callback function for data type and extra data changes.
         * Parameter 0 provides the new data type
         * Parameter 1 provides the new extra data (may be empty)
         * Return True, to accept the change
         */
        using DataTypeChangedCallback = Callback<bool, const std::string&, const Buffer&>;
 
        /**
         * Definition of a callback function for streaming data.
         * Parameter 0 provides the buffer that has been received
         * Parameter 1 provides the size of the buffer, in bytes
         */
        using ReceiveCallback = Callback<void, const void*, const size_t>;
 
        /**
         * Definition of a vector holding channels.
         */
        using Channels = Strings;
 
    public:
 
        /**
         * Creates a new streaming client.
         */
        StreamingClient();
 
        /**
         * Destructs a streaming client.
         */
        ~StreamingClient();
 
        /**
         * Connects the streaming client with a remote streaming server.
         * @param address Address4 of the streaming server
         * @param port Port of the streaming server
         * @return True, if succeeded
         */
        bool connect(const Address4& address, const Port& port);
 
        /**
         * Disconnects the streaming client.
         * @return True, if the client was connected before
         */
        bool disconnect();
 
        /**
         * Returns whether this client is currently connected with a streaming server.
         * @return True, if so
         */
        bool isConnected() const;
 
        /**
         * Returns a list of selectable channels provides by the streaming server.
         * @return Selectable channels
         */
        Channels selectableChannels();
 
        /**
         * Returns a data type of a specified channel provides by the streaming server.
         * @param channel Channel to return the data type for
         * @return Data type
         */
        std::string channelDataType(const std::string& channel);
 
        /**
         * Returns extra data of a specified channel provides by the streaming server.
         * @param channel Channel to return the extra data for
         * @return Extra data (may be empty)
         */
        Buffer channelExtraData(const std::string& channel);
 
        /**
         * Returns the selected channel of the streaming server.
         * @return Selected channel
         */
        inline std::string channel() const;
 
        /**
         * Returns the type of the streaming data.
         * @return Data type
         */
        inline std::string dataType() const;
 
        /**
         * Returns the extra data of the streaming channel.
         * @return Extra data (may be empty)
         */
        inline Buffer extraData() const;
 
        /**
         * Returns the address of the connected streaming server.
         * @return Streaming server address
         */
        inline Address4 serverAddress() const;
 
        /**
         * Returns the port of the connected streaming server.
         * @return Streaming server port
         */
        inline Port serverPort() const;
 
        /**
         * (Re-)Starts the streaming.
         * @param channel Streaming channel, if any
         * @return True, if succeeded
         */
        bool start(const std::string& channel = std::string());
 
        /**
         * Pauses the streaming.
         * @return True, if succeeded
         */
        bool pause();
 
        /**
         * Stops the streaming.
         * @return True, if succeeded
         */
        bool stop();
 
        /**
         * Returns whether this client is currently receiving streaming data.<br>
         * Beware: If the client is paused the client still is handled as in receiving mode.
         * @return True, if so
         */
        inline bool isReceiving();
 
        /**
         * Returns whether this client is currently paused.
         * @return True, if so
         */
        inline bool isPaused();
 
        /**
         * Sets the callback function for streaming server requests (like e.g. start or stop requests).
         * @param callback Callback function to set
         */
        inline void setRequestCallback(const RequestCallback& callback);
 
        /**
         * Sets the callback function for data type and extra data changes from the streaming server.
         * @param callback Callback function to set
         */
        inline void setDataTypeChangedCallback(const DataTypeChangedCallback& callback);
 
        /**
         * Sets the callback function for streaming data received from the streaming server.
         * @param callback Callback function to set
         */
        inline void setReceiveCallback(const ReceiveCallback& callback);
 
    protected:
 
        /**
         * New command function.
         * @param command New command to handle
         * @param value Command value
         * @param sessionId Unique id of the server which is not unique on the client side
         */
        void onCommand(const std::string& command, const std::string& value, const SessionId sessionId);
 
        /**
         * Function handling start commands.
         * @param value Command value
         * @param sessionId Unique id of the server which is not unique on the client side
         */
        void onStart(const std::string& value, const SessionId sessionId);
 
        /**
         * Function handling pause commands.
         * @param value Command value
         * @param sessionId Unique id of the server which is not unique on the client side
         */
        void onPause(const std::string& value, const SessionId sessionId);
 
        /**
         * Function handling stop commands.
         * @param value Command value
         * @param sessionId Unique id of the server which is not unique on the client side
         */
        void onStop(const std::string& value, const SessionId sessionId);
 
        /**
         * Function handling changed data type commands.
         * @param value Command value (format: "dataType|base64EncodedExtraData")
         * @param sessionId Unique id of the server which is not unique on the client side
         */
        void onChangedDataType(const std::string& value, const SessionId sessionId);
 
        /**
         * Callback function for TCP receive message.
         * @param data Received data
         * @param size Size of the received data, in bytes
         */
        void onTCPReceiveData(const void* data, const size_t size);
 
        /**
         * Callback function for UDP receive message.
         * @param address Address4 of the sender
         * @param port Port of the sender
         * @param data Received data
         * @param size Size of the received data
         * @param messageId Unique message id for the received message
         */
        void onUDPReceiveData(const Address4& address, const Port& port, const void* data, const size_t size, const PackagedSocket::MessageId messageId);
 
    protected:
 
        /// Stream channel.
        std::string channel_;
 
        /// Stream data type.
        std::string dataType_;
 
        /// Stream extra data (e.g., codec configuration).
        Buffer extraData_;
 
        /// Determines whether the client is currently receiving.
        bool isReceiving_ = false;
 
        /// Determines whether the client is pause.
        bool isPaused_ = false;
 
        /// UDP client.
        PackagedUDPServer udpClient_;
 
        /// TCP client.
        TCPClient tcpClient_;
 
        /// Address4 of the streaming server.
        Address4 serverAddress_;
 
        /// Port of the streaming server.
        Port serverPort_;
 
        /// Streaming server request callback function.
        RequestCallback requestCallback_;
 
        /// Data type and extra data changed callback function.
        DataTypeChangedCallback dataTypeChangedCallback_;
 
        /// Streaming data receive callback function.
        ReceiveCallback receiveCallback_;
 
        /// Client lock.
        mutable Lock lock_;
};
 
inline std::string StreamingClient::channel() const
{
    const ScopedLock scopedLock(lock_);
    return channel_;
}
 
inline std::string StreamingClient::dataType() const
{
    const ScopedLock scopedLock(lock_);
    return dataType_;
}
 
inline StreamingClient::Buffer StreamingClient::extraData() const
{
    const ScopedLock scopedLock(lock_);
    return extraData_;
}
 
inline bool StreamingClient::isReceiving()
{
    const ScopedLock scopedLock(lock_);
    return isReceiving_;
}
 
inline bool StreamingClient::isPaused()
{
    const ScopedLock scopedLock(lock_);
    return isPaused_;
}
 
inline Address4 StreamingClient::serverAddress() const
{
    const ScopedLock scopedLock(lock_);
    return serverAddress_;
}
 
inline Port StreamingClient::serverPort() const
{
    const ScopedLock scopedLock(lock_);
    return serverPort_;
}
 
inline void StreamingClient::setRequestCallback(const RequestCallback& callback)
{
    const ScopedLock scopedLock(lock_);
    requestCallback_ = callback;
}
 
inline void StreamingClient::setDataTypeChangedCallback(const DataTypeChangedCallback& callback)
{
    const ScopedLock scopedLock(lock_);
    dataTypeChangedCallback_ = callback;
}
 
inline void StreamingClient::setReceiveCallback(const ReceiveCallback& callback)
{
    const ScopedLock scopedLock(lock_);
    receiveCallback_ = callback;
}
 
}
 
}
 
#endif // FACEBOOK_NETWORK_STREAMING_CLIENT_H