Fixed ICE-5196: allow setting an observer with IceMX enabled

3 files changed, 377 insertions, 5 deletions

diff --git a/cpp/test/Ice/metrics/AllTests.cpp b/cpp/test/Ice/metrics/AllTests.cpp
index a181f43fbb1..ea097a349ac 100644
--- a/cpp/test/Ice/metrics/AllTests.cpp
+++ b/cpp/test/Ice/metrics/AllTests.cpp
@@ -9,6 +9,7 @@
 
 #include <Ice/Ice.h>
 #include <TestCommon.h>
+#include <InstrumentationI.h>
 #include <Test.h>
 
 using namespace std;
@@ -368,7 +369,7 @@ toMap(const IceMX::MetricsMap& mmap)
 }
 
 MetricsPrx
-allTests(const Ice::CommunicatorPtr& communicator)
+allTests(const Ice::CommunicatorPtr& communicator, const CommunicatorObserverIPtr& obsv)
 {
     MetricsPrx metrics = MetricsPrx::checkedCast(communicator->stringToProxy("metrics:default -p 12010"));
 
@@ -1047,5 +1048,39 @@ allTests(const Ice::CommunicatorPtr& communicator)
 
     cout << "ok" << endl;
 
+    cout << "testing instrumentation observer delegate... " << flush;
+
+    test(obsv->threadObserver->total > 0);
+    test(obsv->connectionObserver->total > 0);
+    test(obsv->connectionEstablishmentObserver->total > 0);
+    test(obsv->endpointLookupObserver->total > 0);
+    test(obsv->dispatchObserver->total > 0);
+    test(obsv->invocationObserver->total > 0);
+    test(obsv->invocationObserver->remoteObserver->total > 0);
+
+    test(obsv->threadObserver->current > 0);
+    test(obsv->connectionObserver->current > 0);
+    test(obsv->connectionEstablishmentObserver->current == 0);
+    test(obsv->endpointLookupObserver->current == 0);
+    test(obsv->dispatchObserver->current == 0);
+    test(obsv->invocationObserver->current == 0);
+    test(obsv->invocationObserver->remoteObserver->current == 0);
+
+    test(obsv->threadObserver->failedCount == 0);
+    test(obsv->connectionObserver->failedCount > 0);
+    test(obsv->connectionEstablishmentObserver->failedCount > 0);
+    test(obsv->endpointLookupObserver->failedCount > 0);
+    //test(obsv->dispatchObserver->failedCount > 0);
+    test(obsv->invocationObserver->failedCount > 0);
+    test(obsv->invocationObserver->remoteObserver->failedCount > 0);
+
+    test(obsv->threadObserver->states > 0);
+    test(obsv->connectionObserver->received > 0 && obsv->connectionObserver->sent > 0);
+    //test(obsv->dispatchObserver->userExceptionCount > 0);
+    test(obsv->invocationObserver->userExceptionCount > 0 && obsv->invocationObserver->retriedCount > 0);
+    test(obsv->invocationObserver->remoteObserver->replySize > 0);
+
+    cout << "ok" << endl;
+
     return metrics;
 }
diff --git a/cpp/test/Ice/metrics/Client.cpp b/cpp/test/Ice/metrics/Client.cpp
index f30ad6032ad..4f541c2f7cc 100644
--- a/cpp/test/Ice/metrics/Client.cpp
+++ b/cpp/test/Ice/metrics/Client.cpp
@@ -10,6 +10,7 @@
 #include <Ice/Ice.h>
 #include <TestCommon.h>
 #include <Test.h>
+#include <InstrumentationI.h>
 
 DEFINE_TEST("client")
 
@@ -17,10 +18,10 @@ using namespace std;
 using namespace Test;
 
 int
-run(int, char**, const Ice::CommunicatorPtr& communicator)
+run(int, char**, const Ice::CommunicatorPtr& communicator, const CommunicatorObserverIPtr& observer)
 {
-    MetricsPrx allTests(const Ice::CommunicatorPtr&);
-    MetricsPrx metrics = allTests(communicator);
+    MetricsPrx allTests(const Ice::CommunicatorPtr&, const CommunicatorObserverIPtr&);
+    MetricsPrx metrics = allTests(communicator, observer);
     metrics->shutdown();
     return EXIT_SUCCESS;
 }
@@ -40,8 +41,10 @@ main(int argc, char* argv[])
         initData.properties->setProperty("Ice.Admin.DelayCreation", "1");
         initData.properties->setProperty("Ice.Warn.Connections", "0");
         initData.properties->setProperty("Ice.MessageSizeMax", "50000");
+        CommunicatorObserverIPtr observer = new CommunicatorObserverI();
+        initData.observer = observer;
         communicator = Ice::initialize(argc, argv, initData);
-        status = run(argc, argv, communicator);
+        status = run(argc, argv, communicator, observer);
     }
     catch(const Ice::Exception& ex)
     {
diff --git a/cpp/test/Ice/metrics/InstrumentationI.h b/cpp/test/Ice/metrics/InstrumentationI.h
new file mode 100644
index 00000000000..92ff7e1d11d
--- /dev/null
+++ b/cpp/test/Ice/metrics/InstrumentationI.h
@@ -0,0 +1,334 @@
+// **********************************************************************
+//
+// Copyright (c) 2003-2013 ZeroC, Inc. All rights reserved.
+//
+// This copy of Ice is licensed to you under the terms described in the
+// ICE_LICENSE file included in this distribution.
+//
+// **********************************************************************
+
+#ifndef INSTRUMENTATION_I_H
+#define INSTRUMENTATION_I_H
+
+#include <Ice/Instrumentation.h>
+
+class ObserverI : virtual public Ice::Instrumentation::Observer, public IceUtil::Mutex
+{
+public:
+
+    virtual void 
+    reset()
+    {
+        total = 0;
+        current = 0;
+        failedCount = 0;
+    }
+
+    virtual void 
+    attach()
+    {
+        IceUtil::Mutex::Lock sync(*this);
+        ++total;
+        ++current;
+    }
+    virtual void 
+    detach()
+    {
+        IceUtil::Mutex::Lock sync(*this);
+        --current;
+    }
+    virtual void 
+    failed(const std::string&)
+    {
+        IceUtil::Mutex::Lock sync(*this);
+        ++failedCount;
+    }
+
+    Ice::Int total;
+    Ice::Int current;
+    Ice::Int failedCount;
+};
+
+class ConnectionObserverI : public Ice::Instrumentation::ConnectionObserver, public ObserverI
+{
+public:
+
+    virtual void 
+    reset()
+    {
+        IceUtil::Mutex::Lock sync(*this);
+        ObserverI::reset();
+        received = 0;
+        sent = 0;
+    }
+
+    virtual void 
+    sentBytes(Ice::Int s)
+    {
+        IceUtil::Mutex::Lock sync(*this);
+        sent += s;
+    }
+
+    virtual void 
+    receivedBytes(Ice::Int s)
+    {
+        IceUtil::Mutex::Lock sync(*this);
+        received += s;
+    }
+    
+    Ice::Int sent;
+    Ice::Int received;
+};
+
+class ThreadObserverI : public Ice::Instrumentation::ThreadObserver, public ObserverI
+{
+public:
+
+    virtual void 
+    reset()
+    {
+        IceUtil::Mutex::Lock sync(*this);
+        ObserverI::reset();
+        states = 0;
+    }
+
+    virtual void 
+    stateChanged(Ice::Instrumentation::ThreadState, Ice::Instrumentation::ThreadState)
+    {
+        IceUtil::Mutex::Lock sync(*this);
+        ++states;
+    }
+
+    Ice::Int states;
+};
+
+class DispatchObserverI : public Ice::Instrumentation::DispatchObserver, public ObserverI
+{
+public:
+
+    virtual void reset()
+    {
+        IceUtil::Mutex::Lock sync(*this);
+        ObserverI::reset();
+        userExceptionCount = 0;
+        replySize = 0;
+    }
+
+    virtual void 
+    userException()
+    {
+        IceUtil::Mutex::Lock sync(*this);
+        ++userExceptionCount;
+    }
+
+    virtual void 
+    reply(Ice::Int s)
+    {
+        IceUtil::Mutex::Lock sync(*this);
+        replySize += s;
+    }
+
+    Ice::Int userExceptionCount;
+    Ice::Int replySize;
+};
+
+class RemoteObserverI : public Ice::Instrumentation::RemoteObserver, public ObserverI
+{
+public:
+
+    virtual void 
+    reset()
+    {
+        IceUtil::Mutex::Lock sync(*this);
+        ObserverI::reset();
+        replySize = 0;
+    }
+
+    virtual void
+    reply(Ice::Int s)
+    {
+        IceUtil::Mutex::Lock sync(*this);
+        replySize += s;
+    }
+
+    Ice::Int replySize;
+};
+
+class InvocationObserverI : public Ice::Instrumentation::InvocationObserver, public ObserverI
+{
+public:
+
+    virtual void reset()
+    {
+        IceUtil::Mutex::Lock sync(*this);
+        ObserverI::reset();
+        retriedCount = 0;
+        userExceptionCount = 0;
+        if(remoteObserver)
+        {
+            remoteObserver->reset();
+        }
+    }
+
+    virtual void 
+    retried()
+    {
+        IceUtil::Mutex::Lock sync(*this);
+        ++retriedCount;
+    }
+
+    virtual void 
+    userException()
+    {
+        IceUtil::Mutex::Lock sync(*this);
+        ++userExceptionCount;
+    }
+
+    virtual Ice::Instrumentation::RemoteObserverPtr 
+    getRemoteObserver(const Ice::ConnectionInfoPtr& c, const Ice::EndpointPtr& e, Ice::Int, Ice::Int)
+    {
+        IceUtil::Mutex::Lock sync(*this);
+        if(!remoteObserver)
+        {
+            remoteObserver = new RemoteObserverI();
+            remoteObserver->reset();
+        }
+        return remoteObserver;
+    }
+
+    Ice::Int userExceptionCount;
+    Ice::Int retriedCount;
+
+    IceUtil::Handle<RemoteObserverI> remoteObserver;
+};
+
+class CommunicatorObserverI : public Ice::Instrumentation::CommunicatorObserver, public IceUtil::Mutex
+{
+public:
+
+    virtual void 
+    setObserverUpdater(const Ice::Instrumentation::ObserverUpdaterPtr& u)
+    {
+        updater = u;
+    }
+ 
+    virtual Ice::Instrumentation::ObserverPtr 
+    getConnectionEstablishmentObserver(const Ice::EndpointPtr&, const std::string&)
+    {
+        IceUtil::Mutex::Lock sync(*this);
+        if(!connectionEstablishmentObserver)
+        {
+            connectionEstablishmentObserver = new ObserverI();
+            connectionEstablishmentObserver->reset();
+        }
+        return connectionEstablishmentObserver;
+    }
+
+ 
+    virtual Ice::Instrumentation::ObserverPtr 
+    getEndpointLookupObserver(const Ice::EndpointPtr&)
+    {
+        IceUtil::Mutex::Lock sync(*this);
+        if(!endpointLookupObserver)
+        {
+            endpointLookupObserver = new ObserverI();
+            endpointLookupObserver->reset();
+        }
+        return endpointLookupObserver;
+    }
+    
+    virtual Ice::Instrumentation::ConnectionObserverPtr 
+    getConnectionObserver(const Ice::ConnectionInfoPtr&, 
+                          const Ice::EndpointPtr&,
+                          Ice::Instrumentation::ConnectionState, 
+                          const Ice::Instrumentation::ConnectionObserverPtr& old)
+    {
+        IceUtil::Mutex::Lock sync(*this);
+        test(!old || dynamic_cast<ConnectionObserverI*>(old.get()));
+        if(!connectionObserver)
+        {
+            connectionObserver = new ConnectionObserverI();
+            connectionObserver->reset();
+        }
+        return connectionObserver;
+    }
+
+    virtual Ice::Instrumentation::ThreadObserverPtr 
+    getThreadObserver(const std::string&, const std::string&, Ice::Instrumentation::ThreadState,
+                      const Ice::Instrumentation::ThreadObserverPtr& old)
+    {
+        IceUtil::Mutex::Lock sync(*this);
+        test(!old || dynamic_cast<ThreadObserverI*>(old.get()));
+        if(!threadObserver)
+        {
+            threadObserver = new ThreadObserverI();
+            threadObserver->reset();
+        }
+        return threadObserver; 
+   }
+
+    virtual Ice::Instrumentation::InvocationObserverPtr 
+    getInvocationObserver(const Ice::ObjectPrx&, const std::string&, const Ice::Context&)
+    {
+        IceUtil::Mutex::Lock sync(*this);
+        if(!invocationObserver)
+        {
+            invocationObserver = new InvocationObserverI();
+            invocationObserver->reset();
+        }
+        return invocationObserver;
+    }
+
+    virtual Ice::Instrumentation::DispatchObserverPtr 
+    getDispatchObserver(const Ice::Current&, Ice::Int)
+    {
+        IceUtil::Mutex::Lock sync(*this);
+        if(!dispatchObserver)
+        {
+            dispatchObserver = new DispatchObserverI();
+            dispatchObserver->reset();
+        }
+        return dispatchObserver;
+    }
+
+    void reset()
+    {
+        if(connectionEstablishmentObserver)
+        {
+            connectionEstablishmentObserver->reset();
+        }
+        if(endpointLookupObserver)
+        {
+            endpointLookupObserver->reset();
+        }
+        if(connectionObserver)
+        {
+            connectionObserver->reset();
+        }
+        if(threadObserver)
+        {
+            threadObserver->reset();
+        }
+        if(invocationObserver)
+        {
+            invocationObserver->reset();
+        }
+        if(dispatchObserver)
+        {
+            dispatchObserver->reset();
+        }
+    }
+    
+    Ice::Instrumentation::ObserverUpdaterPtr updater;
+
+    IceUtil::Handle<ObserverI> connectionEstablishmentObserver;
+    IceUtil::Handle<ObserverI> endpointLookupObserver;
+    IceUtil::Handle<ConnectionObserverI> connectionObserver;
+    IceUtil::Handle<ThreadObserverI> threadObserver;
+    IceUtil::Handle<InvocationObserverI> invocationObserver;
+    IceUtil::Handle<DispatchObserverI> dispatchObserver;
+};
+
+typedef IceUtil::Handle<CommunicatorObserverI> CommunicatorObserverIPtr;
+
+#endif