AbstractReflectSession.h

/* This file is Copyright 2000-2008 Meyer Sound Laboratories Inc.  See the included LICENSE.txt file for details. */

#ifndef MuscleAbstractReflectSession_h
#define MuscleAbstractReflectSession_h

#include "iogateway/AbstractMessageIOGateway.h"
#include "reflector/AbstractSessionIOPolicy.h"
#include "reflector/ServerComponent.h"
#include "util/Queue.h"
#include "util/RefCount.h"

BEGIN_NAMESPACE(muscle);

class ReflectSessionFactory : public ServerComponent
{
public:
   ReflectSessionFactory();

   virtual ~ReflectSessionFactory() {/* empty */}

   virtual AbstractReflectSessionRef CreateSession(const String & clientAddress, const IPAddressAndPort & factoryInfo) = 0;

   virtual bool IsReadyToAcceptSessions() const {return true;}

   uint32 GetFactoryID() const {return _id;}

protected:
   void BroadcastToAllSessions(const MessageRef & msgRef, void * userData = NULL);

   void BroadcastToAllFactories(const MessageRef & msgRef, void * userData = NULL, bool includeSelf = true);

private:
   uint32 _id;
};

class ProxySessionFactory : public ReflectSessionFactory
{
public:
   ProxySessionFactory(const ReflectSessionFactoryRef & slaveRef) : _slaveRef(slaveRef) {/* empty */}
   virtual ~ProxySessionFactory() {/* empty */}

   virtual status_t AttachedToServer();
   virtual void AboutToDetachFromServer();

   const ReflectSessionFactoryRef & GetSlave() const {return _slaveRef;}

private:
   ReflectSessionFactoryRef _slaveRef;
};

class AbstractReflectSession : public ServerComponent, public AbstractGatewayMessageReceiver
{
public:
   AbstractReflectSession();

   virtual ~AbstractReflectSession();

   const String & GetHostName() const;

   uint16 GetPort() const;

   const ip_address & GetLocalInterfaceAddress() const;

   uint32 GetSessionID() const {return _sessionID;}

   const String & GetSessionIDString() const {return _idString;}

   void EndSession();

   bool DisconnectSession();

   status_t ReplaceSession(const AbstractReflectSessionRef & newSession);

   virtual bool ClientConnectionClosed();

   virtual void AsyncConnectCompleted();

   void SetInputPolicy(const PolicyRef & newPolicy);

   PolicyRef GetInputPolicy() const {return _inputPolicyRef;}

   void SetOutputPolicy(const PolicyRef & newPolicy);

   PolicyRef GetOutputPolicy() const {return _outputPolicyRef;}

   void SetGateway(const AbstractMessageIOGatewayRef & ref) {_gateway = ref; _outputStallLimit = _gateway()?_gateway()->GetOutputStallLimit():MUSCLE_TIME_NEVER;}

   const AbstractMessageIOGatewayRef & GetGateway() const {return _gateway;}

   virtual bool HasBytesToOutput() const;

   virtual bool IsReadyForInput() const;

   virtual int32 DoInput(AbstractGatewayMessageReceiver & receiver, uint32 maxBytes);

   virtual int32 DoOutput(uint32 maxBytes);

   virtual SocketRef CreateDefaultSocket();

   virtual DataIORef CreateDataIO(const SocketRef & socket);

   virtual AbstractMessageIOGatewayRef CreateGateway();

   virtual uint64 GetPulseTime(uint64 now, uint64 sched);

   virtual void Pulse(uint64 now, uint64 sched);

   virtual const char * GetTypeName() const = 0;

   virtual String GetDefaultHostName() const;

   String GetSessionDescriptionString() const;

   const ip_address & GetAsyncConnectIP() const {return _asyncConnectDest.GetIPAddress();}

   uint16 GetAsyncConnectPort() const {return _asyncConnectDest.GetPort();}

   virtual const String & GetSessionRootPath() const {return _sessionRootPath;}

   void SetAutoReconnectDelay(uint64 delay) {_autoReconnectDelay = delay; InvalidatePulseTime();}

   uint64 GetAutoReconnectDelay() const {return _autoReconnectDelay;}

   bool IsConnectingAsync() const {return _connectingAsync;}

   bool IsConnected() const {return _isConnected;}

protected:
   virtual status_t AddOutgoingMessage(const MessageRef & ref);

   void BroadcastToAllSessions(const MessageRef & msgRef, void * userData = NULL, bool includeSelf = true);

   void BroadcastToAllFactories(const MessageRef & msgRef, void * userData = NULL);

   status_t Reconnect();

   const SocketRef & GetSessionSelectSocket() const;

   void SetSessionRootPath(const String & p) {_sessionRootPath = p;}

private:
   void SetPolicyAux(PolicyRef & setRef, uint32 & setChunk, const PolicyRef & newRef, bool isInput);
   void PlanForReconnect();

   friend class ReflectServer;
   uint32 _sessionID;
   String _idString;
   IPAddressAndPort _ipAddressAndPort;
   bool _connectingAsync;
   bool _isConnected;
   String _hostName;
   IPAddressAndPort _asyncConnectDest;
   AbstractMessageIOGatewayRef _gateway;
   uint64 _lastByteOutputAt;
   PolicyRef _inputPolicyRef;
   PolicyRef _outputPolicyRef;
   uint32 _maxInputChunk;   // as determined by our Policy object
   uint32 _maxOutputChunk;  // and stored here for convenience
   uint64 _outputStallLimit;
   bool _scratchReconnected; // scratch, watched by ReflectServer() during ClientConnectionClosed() calls.
   String _sessionRootPath;

   // auto-reconnect support
   uint64 _autoReconnectDelay;
   uint64 _reconnectTime;
   bool _wasConnected;
};

// VC++ (previous to .net) can't handle partial template specialization, so for them we define this explicitly.
#ifdef MUSCLE_USING_OLD_MICROSOFT_COMPILER
DECLARE_HASHTABLE_KEY_CLASS(Ref<AbstractReflectSession>);
#endif

END_NAMESPACE(muscle);

#endif

---