MessageTransceiverThread.h

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

#ifndef MuscleMessageTransceiverThread_h
#define MuscleMessageTransceiverThread_h

#include "system/Thread.h"
#include "reflector/StorageReflectSession.h"
#include "reflector/ReflectServer.h"

BEGIN_NAMESPACE(muscle);

class ThreadSupervisorSession;
class MessageTransceiverThread;

enum {
   MTT_EVENT_INCOMING_MESSAGE = 1835233840, // A new message from a remote computer is ready to process
   MTT_EVENT_SESSION_ACCEPTED,              // A new session has been created by one of our factory objects
   MTT_EVENT_SESSION_ATTACHED,              // A new session has been attached to the local server
   MTT_EVENT_SESSION_CONNECTED,             // A session on the local server has completed its connection to the remote one
   MTT_EVENT_SESSION_DISCONNECTED,          // A session on the local server got disconnected from its remote peer
   MTT_EVENT_SESSION_DETACHED,              // A session on the local server has detached (and been destroyed)
   MTT_EVENT_FACTORY_ATTACHED,              // A ReflectSessionFactory object has been attached to the server
   MTT_EVENT_FACTORY_DETACHED,              // A ReflectSessionFactory object has been detached (and been destroyed)
   MTT_EVENT_OUTPUT_QUEUES_DRAINED,         // Output queues of sessions previously specified in RequestOutputQueuesDrainedNotification() have drained
   MTT_EVENT_SERVER_EXITED,                 // The ReflectServer event loop has terminated
   MTT_LAST_EVENT
};
   
enum {
   MTT_COMMAND_SEND_USER_MESSAGE = 1835230000, // contains a user message to be sent out
   MTT_COMMAND_ADD_NEW_SESSION,                // contains info on a new session to be created
   MTT_COMMAND_PUT_ACCEPT_FACTORY,             // request to start accepting session(s) on a given port
   MTT_COMMAND_REMOVE_ACCEPT_FACTORY,          // remove the acceptor factory on a given port(s)
   MTT_COMMAND_SET_DEFAULT_PATH,               // change the default distribution path
   MTT_COMMAND_NOTIFY_ON_OUTPUT_DRAIN,         // request a notification when all currently pending output has been sent
   MTT_COMMAND_SET_INPUT_POLICY,               // set a new input IO Policy for worker sessions
   MTT_COMMAND_SET_OUTPUT_POLICY,              // set a new output IO Policy for worker sessions
   MTT_COMMAND_REMOVE_SESSIONS,                // remove the matching worker sessions
   MTT_COMMAND_SET_OUTGOING_ENCODING,          // set the MUSCLE_MESSAGE_ENCODING_* setting on worker sessions
   MTT_LAST_COMMAND
};

#define MTT_NAME_PATH        "path"  // field containing a session path (e.g. "/*/*")
#define MTT_NAME_MESSAGE     "mssg"  // field containing a message object
#define MTT_NAME_SOCKET      "sock"  // field containing a Socket reference
#define MTT_NAME_IP_ADDRESS  "addr"  // field containing an int32 IP address
#define MTT_NAME_HOSTNAME    "host"  // field containing an ASCII hostname or IP address
#define MTT_NAME_PORT        "port"  // field containing an int16 port number
#define MTT_NAME_FACTORY_ID  "fcid"  // field containing a uint32 factory ID number (new for v3.40)
#define MTT_NAME_SESSION     "sess"  // field containing an AbstractReflectSession tag
#define MTT_NAME_FROMSESSION "sfrm"  // field containing the root path of the session this message is from (e.g. "192.168.1.103/17")
#define MTT_NAME_FACTORY     "fact"  // field containing a ReflectSessionFactory tag
#define MTT_NAME_DRAIN_TAG   "dtag"  // field containing a DrainTag reference
#define MTT_NAME_POLICY_TAG  "ptag"  // field containing an IOPolicy reference
#define MTT_NAME_ENCODING    "enco"  // field containing the MUSCLE_MESSAGE_ENCODING_* value
#define MTT_NAME_EXPANDLOCALHOST "expl" // boolean field indicating whether localhost IP should be expanded to primary IP
#define MTT_NAME_AUTORECONNECTDELAY "arcd" // int64 indicating how long after disconnect before an auto-reconnect should occur

class DrainTag : public RefCountable
{
public:
   DrainTag() : _notify(NULL) {/* empty */}
 
   virtual ~DrainTag();

private:
   friend class ThreadSupervisorSession;
   friend class ThreadWorkerSession;
   friend class MessageTransceiverThread;

   void SetNotify(ThreadSupervisorSession * notify) {_notify = notify;}
   MessageRef GetReplyMessage() const {return _replyRef;}
   void SetReplyMessage(const MessageRef & ref) {_replyRef = ref;}

   ThreadSupervisorSession * _notify;
   MessageRef _replyRef;
};
typedef Ref<DrainTag> DrainTagRef;

class ThreadWorkerSession : public StorageReflectSession
{
public:
   ThreadWorkerSession();

   virtual ~ThreadWorkerSession();

   virtual status_t AttachedToServer();

   virtual void AboutToDetachFromServer();

   virtual bool ClientConnectionClosed();

   virtual void AsyncConnectCompleted();

   virtual void MessageReceivedFromGateway(const MessageRef & msg, void * userData);

   virtual void MessageReceivedFromSession(AbstractReflectSession & from, const MessageRef & msg, void * userData);

   virtual const char * GetTypeName() const {return "ThreadWorker";}

   virtual int32 DoOutput(uint32 maxBytes);

private:
   friend class ThreadWorkerSessionFactory;
   Queue<DrainTagRef> _drainedNotifiers;
   bool _sendAcceptedMessage;
};
typedef Ref<ThreadWorkerSession> ThreadWorkerSessionRef;

class ThreadWorkerSessionFactory : public StorageReflectSessionFactory
{
public:
   ThreadWorkerSessionFactory();

   virtual status_t AttachedToServer();

   virtual void AboutToDetachFromServer();

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

   virtual ThreadWorkerSessionRef CreateThreadWorkerSession(const String & clientAddress, const IPAddressAndPort & factoryInfo);
};
typedef Ref<ThreadWorkerSessionFactory> ThreadWorkerSessionFactoryRef;

class ThreadSupervisorSession : public StorageReflectSession
{
public:
   ThreadSupervisorSession();

   virtual ~ThreadSupervisorSession();

   virtual void AboutToDetachFromServer();

   virtual AbstractMessageIOGatewayRef CreateGateway();         

   virtual void MessageReceivedFromGateway(const MessageRef & msg, void * userData);

   virtual void MessageReceivedFromSession(AbstractReflectSession & from, const MessageRef & msg, void * userData);

   virtual void MessageReceivedFromFactory(ReflectSessionFactory & from, const MessageRef & msg, void * userData);

   virtual bool ClientConnectionClosed();

   void SetDefaultDistributionPath(const String & path) {_defaultDistributionPath = path;}

   const String & GetDefaultDistributionPath() const {return _defaultDistributionPath;}

   virtual const char * GetTypeName() const {return "ThreadSupervisor";}

protected:
   virtual status_t MessageReceivedFromOwner(const MessageRef & msg, uint32 numLeft);

private:
   friend class MessageTransceiverThread;
   friend class DrainTag;

   void DrainTagIsBeingDeleted(DrainTag * tag);
   void DistributeMessageToWorkers(const MessageRef & distMsg);
   status_t AddNewWorkerConnectSession(const ThreadWorkerSessionRef & sessionRef, const ip_address & hostIP, uint16 port, uint64 autoReconnectDelay);

   Hashtable<DrainTag *, bool> _drainTags;
   String _defaultDistributionPath;
   MessageTransceiverThread * _mtt;
};
typedef Ref<ThreadSupervisorSession> ThreadSupervisorSessionRef;

class MessageTransceiverThread : public Thread
{
public:
   MessageTransceiverThread();

   virtual ~MessageTransceiverThread();

   virtual status_t StartInternalThread();

   virtual status_t SendMessageToSessions(const MessageRef & msgRef, const char * optDistPath = NULL);

   virtual status_t AddNewSession(const SocketRef & socket, const ThreadWorkerSessionRef & optSessionRef);

   status_t AddNewSession(const SocketRef & socket) {return AddNewSession(socket, ThreadWorkerSessionRef());}

   status_t AddNewSession() {return AddNewSession(SocketRef(), ThreadWorkerSessionRef());}

   virtual status_t AddNewConnectSession(const ip_address & targetIPAddress, uint16 port, const ThreadWorkerSessionRef & optSessionRef, uint64 autoReconnectDelay = MUSCLE_TIME_NEVER);

   status_t AddNewConnectSession(const ip_address & targetIPAddress, uint16 port, uint64 autoReconnectDelay = MUSCLE_TIME_NEVER) {return AddNewConnectSession(targetIPAddress, port, ThreadWorkerSessionRef(), autoReconnectDelay);}

   virtual status_t AddNewConnectSession(const String & targetHostName, uint16 port, const ThreadWorkerSessionRef & optSessionRef, bool expandLocalhost = false, uint64 autoReconnectDelay = MUSCLE_TIME_NEVER);

   status_t AddNewConnectSession(const String & targetHostName, uint16 port, bool expandLocalhost = false, uint64 autoReconnectDelay = MUSCLE_TIME_NEVER) {return AddNewConnectSession(targetHostName, port, ThreadWorkerSessionRef(), expandLocalhost, autoReconnectDelay);}

   virtual status_t PutAcceptFactory(uint16 port, const ThreadWorkerSessionFactoryRef & optFactoryRef, const ip_address & optInterfaceIP = invalidIP);

   status_t PutAcceptFactory(uint16 port) {return PutAcceptFactory(port, ThreadWorkerSessionFactoryRef());}

   virtual status_t RemoveAcceptFactory(uint16 port, const ip_address & optInterfaceIP = invalidIP);

   virtual void Reset();

   status_t SetDefaultDistributionPath(const String & distPath);

   const String & GetDefaultDistributionPath() const {return _defaultDistributionPath;}

   int32 GetNextEventFromInternalThread(uint32 & retEventCode, MessageRef * optRetMsgRef = NULL, String * optFromSession = NULL, uint32 * optFromFactoryID = NULL);

   status_t RequestOutputQueuesDrainedNotification(const MessageRef & notificationMsg, const char * optDistPath = NULL, DrainTag * optDrainTag = NULL);

   status_t SetNewInputPolicy(const PolicyRef & pref, const char * optDistPath = NULL);

   status_t SetNewOutputPolicy(const PolicyRef & pref, const char * optDistPath = NULL);

   status_t SetOutgoingMessageEncoding(int32 encoding, const char * optDistPath = NULL);

   status_t RemoveSessions(const char * optDistPath = NULL);

protected:
   virtual void InternalThreadEntry();
 
   virtual ThreadSupervisorSessionRef CreateSupervisorSession();

   virtual ThreadWorkerSessionRef CreateDefaultWorkerSession();

   virtual ThreadWorkerSessionFactoryRef CreateDefaultSessionFactory();

   virtual ReflectServerRef CreateReflectServer();

private:
   friend class ThreadSupervisorSession;
   status_t EnsureServerAllocated();
   status_t SendAddNewSessionMessage(const ThreadWorkerSessionRef & sessionRef, const SocketRef & socket, const char * hostName, const ip_address & hostIP, uint16 port, bool expandLocalhost, uint64 autoReconnectDelay);
   status_t SetNewPolicyAux(uint32 what, const PolicyRef & pref, const char * optDistPath);

   ReflectServerRef _server;
   String _defaultDistributionPath;
};

END_NAMESPACE(muscle);

#endif

---