ReactorInterceptor.h

Go to the documentation of this file.

/*
 *
 *
 * Distributed under the OpenDDS License.
 * See: http://www.opendds.org/license.html
 */

#ifndef OPENDDS_DCPS_REACTORINTERCEPTOR_H
#define OPENDDS_DCPS_REACTORINTERCEPTOR_H

#include "PoolAllocator.h"
#include "PoolAllocationBase.h"
#include "RcEventHandler.h"
#include "dcps_export.h"
#include "unique_ptr.h"
#include "RcHandle_T.h"
#include "ConditionVariable.h"

#include <ace/Reactor.h>
#include <ace/Thread.h>
#include <ace/Thread_Mutex.h>

OPENDDS_BEGIN_VERSIONED_NAMESPACE_DECL

namespace OpenDDS {
namespace DCPS {

class OpenDDS_Dcps_Export ReactorInterceptor : public virtual RcEventHandler {
public:

  class OpenDDS_Dcps_Export Command
  : public virtual RcObject {
  public:
    Command();
    virtual ~Command() { }

    virtual void execute() = 0;

  protected:
    const ACE_Reactor* reactor() const { return reactor_; }
    ACE_Reactor* reactor() { return reactor_; }

  private:
    friend class OpenDDS::DCPS::ReactorInterceptor;
    void set_reactor(ACE_Reactor* reactor) { reactor_ = reactor; }

    ACE_Reactor* reactor_;
  };
  typedef RcHandle<Command> CommandPtr;

  CommandPtr execute_or_enqueue(CommandPtr command);

  virtual bool reactor_is_shut_down() const = 0;

  virtual void reactor(ACE_Reactor *reactor);
  virtual ACE_Reactor* reactor() const;

protected:

  enum ReactorState {
    RS_NONE,
    RS_NOTIFIED,
    RS_PROCESSING
  };

  ReactorInterceptor(ACE_Reactor* reactor,
                     ACE_thread_t owner);

  virtual ~ReactorInterceptor();
  int handle_exception(ACE_HANDLE /*fd*/);
  void process_command_queue_i(ACE_Guard<ACE_Thread_Mutex>& guard);

  ACE_thread_t owner_;
  mutable ACE_Thread_Mutex mutex_;
  typedef OPENDDS_VECTOR(CommandPtr) Queue;
  Queue command_queue_;
  ReactorState state_;
};

typedef RcHandle<ReactorInterceptor> ReactorInterceptor_rch;
typedef WeakRcHandle<ReactorInterceptor> ReactorInterceptor_wrch;

class OpenDDS_Dcps_Export RegisterHandler : public ReactorInterceptor::Command {
public:
  RegisterHandler(ACE_HANDLE io_handle,
                  ACE_Event_Handler* event_handler,
                  ACE_Reactor_Mask mask)
    : io_handle_(io_handle)
    , event_handler_(event_handler)
    , mask_(mask)
  {}

private:
  ACE_HANDLE io_handle_;
  ACE_Event_Handler* event_handler_;
  ACE_Reactor_Mask mask_;

  void execute();
};

class OpenDDS_Dcps_Export RemoveHandler : public ReactorInterceptor::Command {
public:
  RemoveHandler(ACE_HANDLE io_handle,
                ACE_Reactor_Mask mask)
    : io_handle_(io_handle)
    , mask_(mask)
  {}

private:
  ACE_HANDLE io_handle_;
  ACE_Reactor_Mask mask_;

  void execute();
};

} // namespace DCPS
} // namespace OpenDDS

OPENDDS_END_VERSIONED_NAMESPACE_DECL

#endif /* OPENDDS_DCPS_REACTORINTERCEPTOR_H  */