AbstractMessageIOGateway.h

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

#ifndef MuscleAbstractMessageIOGateway_h
#define MuscleAbstractMessageIOGateway_h

#include "dataio/DataIO.h"
#include "message/Message.h"
#include "util/Queue.h"
#include "util/RefCount.h"
#include "util/NetworkUtilityFunctions.h"
#include "util/PulseNode.h"

BEGIN_NAMESPACE(muscle);

class AbstractMessageIOGateway;

class AbstractGatewayMessageReceiver 
{
public:
   AbstractGatewayMessageReceiver() : _inBatch(false), _doInputCount(0) {/* empty */}

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

   void CallMessageReceivedFromGateway(const MessageRef & msg, void * userData = NULL)
   {
      if ((_doInputCount > 0)&&(_inBatch == false))
      {
         _inBatch = true;
         BeginMessageReceivedFromGatewayBatch();
      }
      MessageReceivedFromGateway(msg, userData);
      AfterMessageReceivedFromGateway(msg, userData);
   }

protected:
   virtual void MessageReceivedFromGateway(const MessageRef & msg, void * userData) = 0;

   virtual void AfterMessageReceivedFromGateway(const MessageRef & msg, void * userData) {(void) msg; (void) userData;}

   virtual void BeginMessageReceivedFromGatewayBatch() {/* empty */}

   virtual void EndMessageReceivedFromGatewayBatch() {/* empty */}

private:
   friend class AbstractMessageIOGateway;

   void DoInputBegins() {_doInputCount++;}
   void DoInputEnds()
   {
      if ((--_doInputCount == 0)&&(_inBatch))
      {
         EndMessageReceivedFromGatewayBatch();
         _inBatch = false;
      }
   }

   bool _inBatch;
   uint32 _doInputCount;
};

class QueueGatewayMessageReceiver : public AbstractGatewayMessageReceiver, public Queue<MessageRef>
{
public:
   QueueGatewayMessageReceiver() {/* empty */}

protected:
   virtual void MessageReceivedFromGateway(const MessageRef & msg, void * userData) {(void) userData; (void) AddTail(msg);}
};

class AbstractMessageIOGateway : public RefCountable, public PulseNode
{
public:
   AbstractMessageIOGateway();

   virtual ~AbstractMessageIOGateway();

   status_t AddOutgoingMessage(const MessageRef & messageRef) {return _hosed ? B_ERROR : _outgoingMessages.AddTail(messageRef);}

   int32 DoOutput(uint32 maxBytes = MUSCLE_NO_LIMIT) {return DoOutputImplementation(maxBytes);}

   int32 DoInput(AbstractGatewayMessageReceiver & receiver, uint32 maxBytes = MUSCLE_NO_LIMIT) 
   {
      receiver.DoInputBegins();
      int32 ret = DoInputImplementation(receiver, maxBytes); 
      receiver.DoInputEnds();
      return ret;
   }

   virtual bool IsReadyForInput() const;

   virtual bool HasBytesToOutput() const = 0;

   virtual uint64 GetOutputStallLimit() const;

   virtual void Shutdown();

   virtual void Reset();

   void SetFlushOnEmpty(bool flush);

   bool GetFlushOnEmpty() const {return _flushOnEmpty;}

   Queue<MessageRef> & GetOutgoingMessageQueue() {return _outgoingMessages;}

   const Queue<MessageRef> & GetOutgoingMessageQueue() const {return _outgoingMessages;}

   void SetDataIO(const DataIORef & ref) {_ioRef = ref;}

   const DataIORef & GetDataIO() const {return _ioRef;}

   bool IsHosed() const {return _hosed;}

   virtual status_t ExecuteSynchronousMessaging(AbstractGatewayMessageReceiver * optReceiver, uint64 timeoutPeriod = MUSCLE_TIME_NEVER);

protected:
   virtual int32 DoOutputImplementation(uint32 maxBytes = MUSCLE_NO_LIMIT) = 0;

   virtual int32 DoInputImplementation(AbstractGatewayMessageReceiver & receiver, uint32 maxBytes = MUSCLE_NO_LIMIT) = 0;

   status_t EnsureBufferSize(uint8 ** bufPtr, uint32 * bufSize, uint32 desiredSize, uint32 copySize);

   void FreeLargeBuffer(uint8 ** bufPtr, uint32 * bufSize);

   void SetHosed() {_hosed = true;}
  
protected:
   virtual bool IsStillAwaitingSynchronousMessagingReply() const {return HasBytesToOutput();}

   virtual void SynchronousMessageReceivedFromGateway(const MessageRef & msg, void * userData, AbstractGatewayMessageReceiver & r) {r.MessageReceivedFromGateway(msg, userData);}

   virtual void SynchronousAfterMessageReceivedFromGateway(const MessageRef & msg, void * userData, AbstractGatewayMessageReceiver & r) {r.AfterMessageReceivedFromGateway(msg, userData);}

   virtual void SynchronousBeginMessageReceivedFromGatewayBatch(AbstractGatewayMessageReceiver & r) {r.BeginMessageReceivedFromGatewayBatch();}

   virtual void SynchronousEndMessageReceivedFromGatewayBatch(AbstractGatewayMessageReceiver & r) {r.EndMessageReceivedFromGatewayBatch();}

private:
   friend class ScratchProxyReceiver;
   Queue<MessageRef> _outgoingMessages;

   DataIORef _ioRef;

   bool _hosed;  // set true on error
   bool _flushOnEmpty;

   friend class ReflectServer;
};

typedef Ref<AbstractMessageIOGateway> AbstractMessageIOGatewayRef;

END_NAMESPACE(muscle);

#endif

---