DataDistributionManager.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(DATADISTRIBUTIONMANAGER_H__INCLUDED_)
#define DATADISTRIBUTIONMANAGER_H__INCLUDED_
 
#if _MSC_VER > 1000
#pragma once
#endif // _MSC_VER > 1000
 
#include <string>
#include <map>
#include "DataDistributionManagerTypes.h"
 
typedef const char *(FUNCALL *dataDistributionConfigurationCb)(const void *opaque, const char *channelName, const char *key, const char *value);
typedef void(FUNCALL *dataDistributionLoggingCb)(const void *opaque, const DDM_LOG_LEVEL level, const char *source, const char *function, const char *logStr);
typedef void(FUNCALL *dataDistributionCompletelyDisconnectedCb)(const void *opaque, const char *channelName, const char *reason);
 
struct ClusterHealthElement
{
    int64_t ServerId;          
    DDM_INSTANCE_STATE Status; 
    int64_t Uptime;            
    int64_t LastContactTime;   
};
 
/* Forward declarations */
class IDataDistributionMastershipCallback;
class IDataDistributionMastershipCommon;
class IDataDistributionSubsystem;
 
class DDM_EXPORT ITimeMeasureWrapper
{
public:
    virtual void ResetTime() = 0;
    virtual int64_t ElapsedMilliseconds() = 0;
    virtual int64_t ElapsedMicroseconds() = 0;
    virtual int64_t ElapsedNanoseconds() = 0;
    virtual void AddNanoseconds(unsigned int time) = 0;
};
 
class DDM_EXPORT TimeMeasureWrapper
{
public:
    static ITimeMeasureWrapper *create();
};
 
class DDM_EXPORT IDataDistributionLog
{
public:
    virtual void Log(const DDM_LOG_LEVEL level, const char *sourceName, const char *function, const char *format, ...) = 0;
    virtual void Log(const DDM_LOG_LEVEL level, const char *sourceName, const char *function, const char *format, va_list args) = 0;
};
 
class DDM_EXPORT IDataDistributionCallback
{
public:
    virtual const char *OnConfiguration(const char *channelName, const char *key, const char *value) = 0;
    virtual void OnLogging(DDM_LOG_LEVEL level, const char *source, const char *function, const char *logStr) = 0;
    virtual void OnCompletelyDisconnected(const char *channelName, const char *reason) = 0;
};
 
class DDM_EXPORT DataDistributionCallback : public IDataDistributionCallback
{
public:
    static IDataDistributionCallback *create(void *opaque, dataDistributionConfigurationCb confCb, dataDistributionLoggingCb logCb, dataDistributionCompletelyDisconnectedCb cbCb);
};
 
typedef struct DDM_EXPORT UnderlyingEventData
{
    const char *ChannelName;            
    BOOL IsDataAvailable;               
    int64_t Offset;                     
    int64_t Timestamp;                  
    const char *Key;                    
    size_t KeyLen;                      
    const void *Buffer;                 
    size_t BufferLength;                
    OPERATION_RESULT Condition;         
    int NativeCode;                     
    const char *SubSystemReason;        
    UnderlyingEventData(const char *channelName)
    {
        ChannelName = channelName;
        Condition = DDM_NO_ERROR_CONDITION;
        IsDataAvailable = FALSE;
        Offset = -1;
        Timestamp = -1;
        Key = NULL;
        KeyLen = 0;
        Buffer = NULL;
        BufferLength = 0;
        NativeCode = 0;
        SubSystemReason = NULL;
    }
    UnderlyingEventData(const char *channelName, const char *key, size_t keyLen, const void *buffer, size_t bufferLength, int64_t timestamp = -1, int64_t offset = -1) : UnderlyingEventData(channelName)
    {
        Offset = offset;
        Timestamp = timestamp;
        Key = key;
        KeyLen = keyLen;
        Buffer = buffer;
        BufferLength = bufferLength;
        IsDataAvailable = TRUE;
    }
    UnderlyingEventData(const char *channelName, OPERATION_RESULT condition, int nativeCode, const char *subSystemReason) : UnderlyingEventData(channelName)
    {
        Condition = condition;
        NativeCode = nativeCode;
        SubSystemReason = subSystemReason;
    }
} UnderlyingEventData, *pUnderlyingEventData;
 
typedef void(FUNCALL *dataDistributionUnderlyingEvent)(const void *opaque, const CHANNEL_HANDLE_PARAMETER, const UnderlyingEventData *uEvent);
 
class DDM_EXPORT IDataDistributionChannel
{
public:
    virtual const char* GetChannelName() = 0;
    virtual GENERIC_HANDLE GetOpaqueHandle() = 0;
};
 
class DDM_EXPORT IDataDistributionChannelCallback
{
public:
    virtual void OnUnderlyingEvent(const CHANNEL_HANDLE_PARAMETER, const UnderlyingEventData *uEvent) = 0;
};
 
class DDM_EXPORT DataDistributionChannelCallback
{
public:
    static IDataDistributionChannelCallback *create(void *opaque, dataDistributionUnderlyingEvent uEventCb);
};
 
typedef void(FUNCALL *dataDistributionOnClusterStateChange)(void *opaque, DDM_CLUSTEREVENT change, int64_t serverid);
typedef void(FUNCALL *dataDistributionOnStateChange)(void *opaque, DDM_INSTANCE_STATE newState, DDM_INSTANCE_STATE oldState);
typedef void(FUNCALL *dataDistributionOnStateReady)(void *opaque, void *pState, int64_t len);
typedef void(FUNCALL *dataDistributionOnRequestedState)(void *opaque, void **pState, size_t *len);
typedef void(FUNCALL *dataDistributionOnMultiplePrimary)(void *opaque, int64_t myId, int64_t otherId);
typedef void(FUNCALL *dataDistributionFirstStateChange)(void *opaque, DDM_INSTANCE_STATE newState);
typedef void(FUNCALL *dataDistributionChangingState)(void *opaque, DDM_INSTANCE_STATE actualState, DDM_INSTANCE_STATE futureState);
typedef void(FUNCALL *dataDistributionChangedState)(void *opaque, DDM_INSTANCE_STATE actualState);
 
class DDM_EXPORT IDataDistributionMastershipCallback
{
public:
    virtual void OnClusterStateChange(DDM_CLUSTEREVENT change, int64_t serverid) = 0;
    virtual void OnStateChange(DDM_INSTANCE_STATE newState, DDM_INSTANCE_STATE oldState) = 0;
    virtual void OnStateReady(void *pState, int64_t len) = 0;
    virtual void OnRequestedState(void **pState, size_t *len) = 0;
    virtual void OnMultiplePrimary(int64_t myId, int64_t otherId) = 0;
    virtual void FirstStateChange(DDM_INSTANCE_STATE newState) = 0;
    virtual void ChangingState(DDM_INSTANCE_STATE actualState, DDM_INSTANCE_STATE futureState) = 0;
    virtual void ChangedState(DDM_INSTANCE_STATE actualState) = 0;
};
class DDM_EXPORT DataDistributionMastershipCallback
{
public:
    static IDataDistributionMastershipCallback *create(void *opaque, dataDistributionOnClusterStateChange c1, dataDistributionOnStateChange c2,
                                                       dataDistributionOnStateReady c3, dataDistributionOnRequestedState c4, dataDistributionOnMultiplePrimary c5,
                                                       dataDistributionFirstStateChange c6, dataDistributionChangingState c7, dataDistributionChangedState c8);
};
class DDM_EXPORT IDataDistributionChannelBase
{
public:
    virtual CHANNEL_HANDLE CreateChannel(const char *channelName, IDataDistributionChannelCallback *dataCb, DDM_CHANNEL_DIRECTION direction = DDM_CHANNEL_DIRECTION::ALL, const char *arrayParams[] = NULL, int len = 0) = 0;
    virtual OPERATION_RESULT StartChannel(CHANNEL_HANDLE_PARAMETER, unsigned long timeout) = 0;
    virtual OPERATION_RESULT StopChannel(CHANNEL_HANDLE_PARAMETER, unsigned long timeout) = 0;
    virtual void SetParameter(CHANNEL_HANDLE_PARAMETER, const char *paramName, const char *paramValue) = 0;
    virtual void SetParameter(CHANNEL_HANDLE_PARAMETER, DDM_GENERAL_PARAMETER paramId, const char *paramValue) = 0;
    virtual const char *GetParameter(CHANNEL_HANDLE_PARAMETER, const char *paramName) = 0;
    virtual const char *GetParameter(CHANNEL_HANDLE_PARAMETER, DDM_GENERAL_PARAMETER paramId) = 0;
    virtual OPERATION_RESULT Lock(CHANNEL_HANDLE_PARAMETER, unsigned long timeout) = 0;
    virtual OPERATION_RESULT Unlock(CHANNEL_HANDLE_PARAMETER) = 0;
    virtual OPERATION_RESULT SeekChannel(CHANNEL_HANDLE_PARAMETER, int64_t position, DDM_SEEKCONTEXT context = DDM_SEEKCONTEXT::OFFSET, DDM_SEEKKIND kind = DDM_SEEKKIND::ABSOLUTE) = 0;
    virtual OPERATION_RESULT DeleteChannel(CHANNEL_HANDLE_PARAMETER) = 0;
    virtual OPERATION_RESULT WriteOnChannel(CHANNEL_HANDLE_PARAMETER, const char *key, size_t keyLen, void *buffer, size_t bufferLen, const BOOL waitAll = FALSE, const int64_t timestamp = DDM_NO_TIMESTAMP) = 0;
    virtual OPERATION_RESULT ReadFromChannel(CHANNEL_HANDLE_PARAMETER, int64_t offset, size_t *bufferLen, void **buffer) = 0;
    virtual OPERATION_RESULT ChangeChannelDirection(CHANNEL_HANDLE_PARAMETER, DDM_CHANNEL_DIRECTION direction) = 0;
};
class DDM_EXPORT IDataDistributionTransport : public IDataDistributionChannelBase
{
public:
    virtual int GetServerLostTimeout() = 0;
    virtual size_t GetMaxMessageSize() = 0;
    virtual OPERATION_RESULT Start(unsigned long timeout) = 0;
    virtual OPERATION_RESULT Stop(unsigned long timeout) = 0;
};
class DDM_EXPORT IDataDistributionSubsystem : public IDataDistributionTransport, public IDataDistributionLog
{
public:
    virtual OPERATION_RESULT Initialize(IDataDistributionCallback *cb, const char *conf_file, const char *hostAddress = NULL, const char *channelTrailer = NULL) = 0;
    virtual OPERATION_RESULT Initialize(IDataDistributionCallback *cb, const char *arrayParams[], int len, const char *hostAddress = NULL, const char *channelTrailer = NULL) = 0;
    virtual IDataDistributionCallback *GetCallbacks() = 0;
};
typedef std::map<int64_t, ClusterHealthElement *> ClusterHealth;
typedef std::map<int64_t, ClusterHealthElement *>::const_iterator ClusterHealthIterator;
 
class DDM_EXPORT IDataDistributionMastershipCommon
{
public:
    virtual OPERATION_RESULT Initialize(IDataDistributionSubsystem *transportManager, IDataDistributionMastershipCallback *cbs, const char *hostAddress = NULL, const char *arrayParams[] = NULL, int len = 0) = 0;
    virtual void SetParameter(const char *paramName, const char *paramValue) = 0;
    virtual const char *GetParameter(const char *paramName) = 0;
    virtual OPERATION_RESULT Start(unsigned long timeout) = 0;
    virtual OPERATION_RESULT Stop(unsigned long timeout) = 0;
    virtual BOOL GetIamNextPrimary() = 0;
    virtual BOOL RequestIAmNextPrimary() = 0;
    virtual int64_t *GetClusterIndexes(size_t *length) = 0;
    virtual DDM_INSTANCE_STATE GetStateOf(int64_t serverId) = 0;
    virtual ClusterHealth GetClusterHealth() = 0;
    virtual DDM_INSTANCE_STATE GetMyState() = 0;
    virtual void ChangeMyState(DDM_INSTANCE_STATE state) = 0;
    virtual void ChangeState(int64_t instanceId, DDM_INSTANCE_STATE state) = 0;
    virtual int64_t GetLocalServerId() = 0;
    virtual int64_t GetPrimaryServerId() = 0;
    virtual int64_t GetMessageDelay() = 0;
    virtual int64_t GetUpTime() = 0;
};
 
class DDM_EXPORT IDataDistribution
{
public:
    virtual OPERATION_RESULT Initialize(IDataDistributionCallback *iddcb, const char *conf_file = NULL, const char *hostAddress = NULL, const char *channelTrailer = NULL) = 0;
    virtual OPERATION_RESULT Initialize(IDataDistributionCallback *iddcb, const char *arrayParams[], int len, const char *hostAddress = NULL, const char *channelTrailer = NULL) = 0;
    virtual OPERATION_RESULT RequestMastershipManager(IDataDistributionMastershipCallback *cbs, const char *szMyAddress = NULL, const char *arrayParams[] = NULL, int len = 0) = 0;
    virtual BOOL Start(unsigned long timeout) = 0;
    virtual BOOL Stop(unsigned long timeout) = 0;
    virtual IDataDistributionSubsystem *GetSubsystemManager() = 0;
    virtual IDataDistributionMastershipCommon *GetMastershipManager() = 0;
    virtual std::string GetProtocol() = 0;
    virtual std::string GetProtocolLib() = 0;
    virtual std::string GetMastershipLib() = 0;
};
 
class DDM_EXPORT DataDistribution : public IDataDistribution
{
public:
    static IDataDistribution *create();
};
 
#endif // !defined(DATADISTRIBUTIONMANAGER_H__INCLUDED_)