SmartDataDistributionManager.h

Go to the documentation of this file.

/*
*  Copyright 2023 MASES s.r.l.
*
*  Licensed under the Apache License, Version 2.0 (the "License");
*  you may not use this file except in compliance with the License.
*  You may obtain a copy of the License at
*
*  http://www.apache.org/licenses/LICENSE-2.0
*
*  Unless required by applicable law or agreed to in writing, software
*  distributed under the License is distributed on an "AS IS" BASIS,
*  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
*  See the License for the specific language governing permissions and
*  limitations under the License.
*
*  Refer to LICENSE for more information.
*/
 
#if !defined(SMARTDATADISTRIBUTIONMANAGER_H__INCLUDED_)
#define SMARTDATADISTRIBUTIONMANAGER_H__INCLUDED_
 
#include "DataDistributionManager.h"
class DDM_EXPORT SmartTimeMeasureWrapper : public ITimeMeasureWrapper
{
public:
    SmartTimeMeasureWrapper();
    ~SmartTimeMeasureWrapper();
    void ResetTime();
    int64_t ElapsedMilliseconds();
    int64_t ElapsedMicroseconds();
    int64_t ElapsedNanoseconds();
    void AddNanoseconds(unsigned int time);
 
private:
    ITimeMeasureWrapper *m_pITimeMeasureWrapper;
};
 
class DDM_EXPORT SmartDataDistributionChannel : protected IDataDistributionChannelCallback
{
public:
    SmartDataDistributionChannel();
 
protected:
    ~SmartDataDistributionChannel();
 
public:
    DDM_CHANNEL_DIRECTION GetDirection();
    void SetInformation(const char *channelName, IDataDistributionChannelBase *pIDataDistributionChannelBase, CHANNEL_HANDLE_PARAMETER, DDM_CHANNEL_DIRECTION direction);
    OPERATION_RESULT StartChannel(unsigned long timeout);
    OPERATION_RESULT StopChannel(unsigned long timeout);
    OPERATION_RESULT Lock(unsigned long timeout);
    OPERATION_RESULT Unlock();
    OPERATION_RESULT SeekChannel(int64_t position);
    OPERATION_RESULT WriteOnChannel(const char *key, size_t keyLen, void *buffer, size_t bufferLen, const BOOL waitAll = FALSE, const int64_t timestamp = DDM_NO_TIMESTAMP);
    OPERATION_RESULT ReadFromChannel(int64_t offset, size_t *dataLen, void **buffer);
    OPERATION_RESULT ChangeChannelDirection(DDM_CHANNEL_DIRECTION direction);
    const char *GetChannelName();
    int64_t GetTimestamp();
    int64_t GetOffset();
    virtual void OnDataAvailable(const char *key, size_t keyLen, const void *buffer, const size_t len);
    virtual void OnDataAvailable(const std::string key, const void *buffer, const size_t len);
    virtual void OnConditionOrError(const OPERATION_RESULT errorCode, const int nativeCode, const char *subSystemReason);
    virtual void OnConditionOrError(const OPERATION_RESULT errorCode, const int nativeCode, const std::string subSystemReason);
 
protected:
    virtual void OnUnderlyingEvent(const CHANNEL_HANDLE_PARAMETER, const UnderlyingEventData *uEvent);
 
private:
    const char *m_ChannelName;
    int64_t m_timestamp;
    int64_t m_offset;
    CHANNEL_HANDLE m_channelHandle;
    IDataDistributionChannelBase *m_pIDataDistributionChannelBase;
    DDM_CHANNEL_DIRECTION m_Direction;
};
 
class DDM_EXPORT SmartDataDistributionBase : protected IDataDistributionCallback, protected IDataDistributionMastershipCallback
{
public:
    SmartDataDistributionBase();
    ~SmartDataDistributionBase();
    OPERATION_RESULT Initialize(const char *conf_file, const char *hostAddress = 0, const char *channelTrailer = 0);
    OPERATION_RESULT Initialize(const char *arrayParams[], int len, const char *hostAddress = 0, const char *channelTrailer = 0);
    OPERATION_RESULT RequestMastershipManager(const char *hostAddress = NULL, const char *arrayParams[] = NULL, int len = 0);
    BOOL Start(unsigned long timeout);
    BOOL Stop(unsigned long timeout);
    std::string GetProtocol();
    std::string GetProtocolLib();
    std::string GetMastershipLib();
 
protected:
    // IDataDistributionCallback
    virtual const char *OnConfiguration(const char *channelName, const char *key, const char *value);
    virtual void OnLogging(DDM_LOG_LEVEL level, const char *source, const char *function, const char *logStr);
    virtual void OnCompletelyDisconnected(const char *channelName, const char *reason);
    // IDataDistributionMastershipCallback
    virtual void OnClusterStateChange(DDM_CLUSTEREVENT change, int64_t serverid);
    virtual void OnStateChange(DDM_INSTANCE_STATE newState, DDM_INSTANCE_STATE oldState);
    virtual void OnStateReady(void *pState, int64_t len);
    virtual void OnRequestedState(void **pState, size_t *len);
    virtual void OnMultiplePrimary(int64_t myId, int64_t otherId);
    virtual void FirstStateChange(DDM_INSTANCE_STATE newState);
    virtual void ChangingState(DDM_INSTANCE_STATE actualState, DDM_INSTANCE_STATE futureState);
    virtual void ChangedState(DDM_INSTANCE_STATE actualState);
 
protected:
    IDataDistribution *m_pIDataDistribution;
    IDataDistributionMastershipCommon *m_pIDataDistributionMastershipCommon;
    OPERATION_RESULT m_pInitializeHRESULT;
};
 
template <typename T>
class SmartDataDistribution : public SmartDataDistributionBase
{
public:
    SmartDataDistribution() : SmartDataDistributionBase() {}
    ~SmartDataDistribution() {}
    T *CreateSmartChannel(const char *channelName, DDM_CHANNEL_DIRECTION direction = DDM_CHANNEL_DIRECTION::ALL, const char *arrayParams[] = NULL, int len = 0)
    {
        static_assert(std::is_base_of<SmartDataDistributionChannel, T>::value, "type parameter of this class must derive from SmartDataDistributionChannel");
        if (OPERATION_FAILED(m_pInitializeHRESULT))
            return NULL;
        T *pSmartChannelT = new T();
        SmartDataDistributionChannel *pSmartChannel = (SmartDataDistributionChannel *)pSmartChannelT;
        IDataDistributionChannelBase *pChannelBase = m_pIDataDistribution->GetSubsystemManager();
        CHANNEL_HANDLE_PARAMETER = pChannelBase->CreateChannel(channelName, (IDataDistributionChannelCallback *)pSmartChannel, direction, arrayParams, len);
        pSmartChannel->SetInformation(channelName, pChannelBase, channelHandle, direction);
 
        return pSmartChannelT;
    }
};
 
#endif // End SMARTDATADISTRIBUTIONMANAGER_H__INCLUDED_