Removed thread-per-connection and added serialize mode

1 files changed, 12 insertions, 6 deletions

diff --git a/cpp/src/IceStorm/Subscriber.cpp b/cpp/src/IceStorm/Subscriber.cpp
index 0b7338c321f..632febd274b 100644
--- a/cpp/src/IceStorm/Subscriber.cpp
+++ b/cpp/src/IceStorm/Subscriber.cpp
@@ -168,7 +168,7 @@ private:
     const SubscriberPtr _subscriber;
 };
 
-class OnewayIceInvokeI : public Ice::AMI_Object_ice_invoke
+class OnewayIceInvokeI : public Ice::AMI_Object_ice_invoke, public Ice::AMISentCallback
 {
 public:
 
@@ -183,9 +183,9 @@ public:
         assert(false);
     }
 
-    virtual void __sent(Ice::ConnectionI* c)
+    virtual void
+    ice_sent()
     {
-        AMI_Object_ice_invoke::__sent(c);
         _subscriber->sent();
     }
 
@@ -365,11 +365,12 @@ SubscriberOneway::flush()
         //
         EventDataPtr e = _events.front();
         _events.erase(_events.begin());
-        ++_outstanding;
-          
         try
         {
-            _obj->ice_invoke_async(new OnewayIceInvokeI(this), e->op, e->mode, e->data, e->context);
+            if(!_obj->ice_invoke_async(new OnewayIceInvokeI(this), e->op, e->mode, e->data, e->context))
+            {
+                ++_outstanding;
+            }
         }
         catch(const Ice::Exception& ex)
         {
@@ -377,6 +378,11 @@ SubscriberOneway::flush()
             return;
         }
     }
+
+    if(_events.empty() && _outstanding == 0 && _shutdown)
+    {
+        _lock.notify();
+    }
 }
 
 void