ProxyPushSupplier.cc Source File

00001 //                            Package   : omniEvents
00002 // ProxyPushSupplier.cc       Created   : 2003/12/04
00003 //                            Author    : Alex Tingle
00004 //
00005 //    Copyright (C) 2003 Alex Tingle.
00006 //
00007 //    This file is part of the omniEvents application.
00008 //
00009 //    omniEvents is free software; you can redistribute it and/or
00010 //    modify it under the terms of the GNU Lesser General Public
00011 //    License as published by the Free Software Foundation; either
00012 //    version 2.1 of the License, or (at your option) any later version.
00013 //
00014 //    omniEvents is distributed in the hope that it will be useful,
00015 //    but WITHOUT ANY WARRANTY; without even the implied warranty of
00016 //    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
00017 //    Lesser General Public License for more details.
00018 //
00019 //    You should have received a copy of the GNU Lesser General Public
00020 //    License along with this library; if not, write to the Free Software
00021 //    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
00022 //
00023 
00024 #include "ProxyPushSupplier.h"
00025 #include "Orb.h"
00026 #include "omniEventsLog.h"
00027 #include "PersistNode.h"
00028 #include <assert.h>
00029 
00030 namespace OmniEvents {
00031 
00035 class omni_mutex_kcol {
00036     omni_mutex& mutex;
00037 public:
00038     omni_mutex_kcol(omni_mutex& m) : mutex(m) { mutex.unlock(); }
00039     ~omni_mutex_kcol(void) { mutex.lock(); }
00040 private:
00041     // dummy copy constructor and operator= to prevent copying
00042     omni_mutex_kcol(const omni_mutex_kcol&);
00043     omni_mutex_kcol& operator=(const omni_mutex_kcol&);
00044 };
00045 
00046 
00047 //
00048 //  ProxyPushSupplierManager
00049 //
00050 
00051 PortableServer::Servant
00052 ProxyPushSupplierManager::incarnate(
00053   const PortableServer::ObjectId& oid,
00054   PortableServer::POA_ptr         poa
00055 )
00056 {
00057   ProxyPushSupplier_i* result =new ProxyPushSupplier_i(_managedPoa,_queue);
00058   PauseThenWake p(this);
00059   _servants.insert(result);
00060   return result;
00061 }
00062 
00063 void
00064 ProxyPushSupplierManager::etherealize(
00065   const PortableServer::ObjectId& oid,
00066   PortableServer::POA_ptr         adapter,
00067   PortableServer::Servant         serv,
00068   CORBA::Boolean                  cleanup_in_progress,
00069   CORBA::Boolean                  remaining_activations
00070 )
00071 {
00072   omni_mutex_lock pause(_lock);
00073   ProxyManager::etherealize(oid,adapter,serv,
00074     cleanup_in_progress,remaining_activations);
00075 }
00076 
00077 ProxyPushSupplierManager::ProxyPushSupplierManager(
00078   PortableServer::POA_ptr parentPoa,
00079   EventQueue& q
00080 )
00081 : ProxyManager(parentPoa,"ProxyPushSupplier"),
00082   omni_thread(NULL,PRIORITY_HIGH),
00083   _queue(q),
00084   _lock(),_condition(&_lock),
00085   _refCount(1)
0008he terms of the GNU Lesser General Public
00011 //    License as published by the Free Software Foundation; either
00012 //    version 2.1 of the License, or (at your option) any later version.
00013 //
00014 //    omniEvents is distributed in the hope that it will be useful,
00015 //    but WITHOUT ANY WARRANTY; without even the implied warranty of
00016 //    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
00017 //    Lesser General Public License for more details.
00018 //
00019 //    You should have received a copy of the GNU Lesser General Public
00020 //    License along with this library; if not, write to the Free Software
00021 //    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
00022 //
00023 
00024 #ifndef OMNIEVENTS__PROXYPUSHCONSUMER_H
00025 #define OMNIEVENTS__PROXYPUSHCONSUMER_H
00026 
00027 #ifdef HAVE_CONFIG_H
00028 #  include "config.h"
00029 #endif
00030 
00031 #include <string>
00032 #include <map>
00033 #include <list>
00034 
00035 #ifdef HAVE_IOSTREAM
00036 #  include <iostream>
00037 #else
00038 #  include <iostream.h>
00039 #endif
00040 
00041 #include "Callback.h"
00042 #include "Servant.h"
00043 
00044 #include "CosEventChannelAdmin.hh"
00045 
00046 #ifdef HAVE_STD_IOSTREAM
00047 using namespace std;
00048 #endif
00049 
00050 namespace OmniEvents {
00051 
00052 class PersistNode;
00053 class ConsumerAdmin_i;
00054 
00060 class ProxyPushConsumer_i
00061 : public virtual POA_CosEventChannelAdmin::ProxyPushConsumer,
00062   public Servant
00063 {
00064 public: // CORBA interface methods
00068   void connect_push_supplier(CosEventComm::PushSupplier_ptr pushSupplier);
00069 
00073   void disconnect_push_consumer();
00074 
00076   void push(const CORBA::Any& event);
00077 
00078 public:
00079   ProxyPushConsumer_i(
00080     PortableServer::POA_ptr parentPoa,
00081     list<CORBA::Any*>&      q,
00082     ConsumerAdmin_i&        consumerAdmin
00083   );
00084   virtual ~ProxyPushConsumer_i();
00085   
00086   void trigger() {_useLocalQueue=false;}
00087 
00094   CosEventChannelAdmin::ProxyPushConsumer_ptr createObject();
00095 
00097   void disconnect();
00098 
00100   void reincarnate(const PersistNode& node);
00102   void output(ostream& os) const;
00103 
00104 private:
00105   string currentObjectId() const;
00106   struct Connection : public Callback
00107   {
00108     const char*                    _channelName;
00109     string                         _oidstr;
00110     CosEventComm::PushSupplier_var _target;
00111     bool _targetIsProxy; 
00112 
00114     Connection(
00115       const char*                    channelName,
00116       const string&                  oidstr,
00117       CosEventComm::PushSupplier_ptr pushSupplier,
00118       bool                           isProxy=false
00119     );
00120     virtual ~Connection(){}
00122     void callback(CORBA::Request_ptr req);
00124     void output(ostream& os) const;
00125   private:
00126     Connection(); 
00127   };
00128 
00129   typedef map<string,Connection> Connections_t;
00130   Connections_t      _connections;
00131   CORBA::String_var  _channelName;
00132   ConsumerAdmin_i&   _consumerAdmin;
00133   list<CORBA::Any*>& _queue;
00134   bool               _useLocalQueue; 
00135 };
00136 
00137 }; // end namespace OmniEvents
00138 
00139 #endif // OMNIEVENTS__PROXYPUSHCONSUMER_H

00001 //                            Package   : omniEvents
00002 // ProxyPushSupplier.cc       Created   : 2003/12/04
00003 //                            Author    : Alex Tingle
00004 //
00005 //    Copyright (C) 2003 Alex Tingle.
00006 //
00007 //    This file is part of the omniEvents application.
00008 //
00009 //    omniEvents is free software; you can redistribute it and/or
00010 //    modify it under the terms of the GNU Lesser General Public
00011 //    License as published by the Free Software Foundation; either
00012 //    version 2.1 of the License, or (at your option) any later version.
00013 //
00014 //    omniEvents is distributed in the hope that it will be useful,
00015 //    but WITHOUT ANY WARRANTY; without even the implied warranty of
00016 //    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
00017 //    Lesser General Public License for more details.
00018 //
00019 //    You should have received a copy of the GNU Lesser General Public
00020 //    License along with this library; if not, write to the Free Software
00021 //    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
00022 //
00023 
00024 #include "ProxyPushSupplier.h"
00025 #include "Orb.h"
00026 #include "omniEventsLog.h"
00027 #include "PersistNode.h"
00028 #include <assert.h>
00029 
00030 namespace OmniEvents {
00031 
00035 class omni_mutex_kcol {
00036     omni_mutex& mutex;
00037 public:
00038     omni_mutex_kcol(omni_mutex& m) : mutex(m) { mutex.unlock(); }
00039     ~omni_mutex_kcol(void) { mutex.lock(); }
00040 private:
00041     // dummy copy constructor and operator= to prevent copying
00042     omni_mutex_kcol(const omni_mutex_kcol&);
00043     omni_mutex_kcol& operator=(const omni_mutex_kcol&);
00044 };
00045 
00046 
00047 //
00048 //  ProxyPushSupplierManager
00049 //
00050 
00051 PortableServer::Servant
00052 ProxyPushSupplierManager::incarnate(
00053   const PortableServer::ObjectId& oid,
00054   PortableServer::POA_ptr         poa
00055 )
00056 {
00057   ProxyPushSupplier_i* result =new ProxyPushSupplier_i(_managedPoa,_queue);
00058   PauseThenWake p(this);
00059   _servants.insert(result);
00060   return result;
00061 }
00062 
00063 void
00064 ProxyPushSupplierManager::etherealize(
00065   const PortableServer::ObjectId& oid,
00066   PortableServer::POA_ptr         adapter,
00067   PortableServer::Servant         serv,
00068   CORBA::Boolean                  cleanup_in_progress,
00069   CORBA::Boolean                  remaining_activations
00070 )
00071 {
00072   omni_mutex_lock pause(_lock);
00073   ProxyManager::etherealize(oid,adapter,serv,
00074     cleanup_in_progress,remaining_activations);
00075 }
00076 
00077 ProxyPushSupplierManager::ProxyPushSupplierManager(
00078   PortableServer::POA_ptr parentPoa,
00079   EventQueue& q
00080 )
00081 : ProxyManager(parentPoa,"ProxyPushSupplier"),
00082   omni_thread(NULL,PRIORITY_HIGH),
00083   _queue(q),
00084   _lock(),_condition(&_lock),
00085   _refCount(1)
0008he terms of the GNU Lesser General Public
00011 //    License as published by the Free Software Foundation; either
00012 //    version 2.1 of the License, or (at your option) any later version.
00013 //
00014 //    omniEvents is distributed in the hope that it will be useful,
00015 //    but WITHOUT ANY WARRANTY; without even the implied warranty of
00016 //    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
00017 //    Lesser General Public License for more details.
00018 //
00019 //    You should have received a copy of the GNU Lesser General Public
00020 //    License along with this library; if not, write to the Free Software
00021 //    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
00022 //
00023 
00024 #ifndef OMNIEVENTS__PROXYPUSHCONSUMER_H
00025 #define OMNIEVENTS__PROXYPUSHCONSUMER_H
00026 
00027 #ifdef HAVE_CONFIG_H
00028 #  include "config.h"
00029 #endif
00030 
00031 #include <string>
00032 #include <map>
00033 #include <list>
00034 
00035 #ifdef HAVE_IOSTREAM
00036 #  include <iostream>
00037 #else
00038 #  include <iostream.h>
00039 #endif
00040 
00041 #include "Callback.h"
00042 #include "Servant.h"
00043 
00044 #include "CosEventChannelAdmin.hh"
00045 
00046 #ifdef HAVE_STD_IOSTREAM
00047 using namespace std;
00048 #endif
00049 
00050 namespace OmniEvents {
00051 
00052 class PersistNode;
00053 class ConsumerAdmin_i;
00054 
00060 class ProxyPushConsumer_i
00061 : public virtual POA_CosEventChannelAdmin::ProxyPushConsumer,
00062   public Servant
00063 {
00064 public: // CORBA interface methods
00068   void connect_push_supplier(CosEventComm::PushSupplier_ptr pushSupplier);
00069 
00073   void disconnect_push_consumer();
00074 
00076   void push(const CORBA::Any& event);
00077 
00078 public:
00079   ProxyPushConsumer_i(
00080     PortableServer::POA_ptr parentPoa,
00081     list<CORBA::Any*>&      q,
00082     ConsumerAdmin_i&        consumerAdmin
00083   );
00084   virtual ~ProxyPushConsumer_i();
00085   
00086   void trigger() {_useLocalQueue=false;}
00087 
00094   CosEventChannelAdmin::ProxyPushConsumer_ptr createObject();
00095 
00097   void disconnect();
00098 
00100   void reincarnate(const PersistNode& node);
00102   void output(ostream& os) const;
00103 
00104 private:
00105   string currentObjectId() const;
00106   struct Connection : public Callback
00107   {