Python changes:

 - Adding ice_flushBatchRequests, ice_flushBatchRequests_async
 - Adding support for ice_sent callback

6 files changed, 333 insertions, 44 deletions

diff --git a/py/modules/IcePy/Communicator.cpp b/py/modules/IcePy/Communicator.cpp
index 23231208a50..6dea0305af6 100644
--- a/py/modules/IcePy/Communicator.cpp
+++ b/py/modules/IcePy/Communicator.cpp
@@ -141,6 +141,7 @@ communicatorInit(CommunicatorObject* self, PyObject* args, PyObject* /*kwds*/)
         PyObjectHandle properties = PyObject_GetAttrString(initData, STRCAST("properties"));
         PyObjectHandle logger = PyObject_GetAttrString(initData, STRCAST("logger"));
         PyObjectHandle threadHook = PyObject_GetAttrString(initData, STRCAST("threadHook"));
+        PyErr_Clear(); // PyObject_GetAttrString sets an error on failure.
 
         if(properties.get() && properties.get() != Py_None)
         {
@@ -426,14 +427,7 @@ communicatorIsShutdown(CommunicatorObject* self)
         return 0;
     }
 
-    if(isShutdown)
-    {
-        PyRETURN_TRUE;
-    }
-    else
-    {
-        PyRETURN_FALSE;
-    }
+    PyRETURN_BOOL(isShutdown);
 }
 
 #ifdef WIN32
diff --git a/py/modules/IcePy/ImplicitContext.cpp b/py/modules/IcePy/ImplicitContext.cpp
index 039ece7554c..0275180e24d 100644
--- a/py/modules/IcePy/ImplicitContext.cpp
+++ b/py/modules/IcePy/ImplicitContext.cpp
@@ -146,14 +146,7 @@ implicitContextContainsKey(ImplicitContextObject* self, PyObject* args)
         return 0;
     }
 
-    if(containsKey)
-    {
-        PyRETURN_TRUE;
-    }
-    else
-    {
-        PyRETURN_FALSE;
-    }
+    PyRETURN_BOOL(containsKey);
 }
 
 #ifdef WIN32
diff --git a/py/modules/IcePy/Operation.cpp b/py/modules/IcePy/Operation.cpp
index 39340e0f60e..843a9ba44ca 100644
--- a/py/modules/IcePy/Operation.cpp
+++ b/py/modules/IcePy/Operation.cpp
@@ -144,11 +144,23 @@ public:
     virtual void ice_response(bool, const pair<const Ice::Byte*, const Ice::Byte*>&);
     virtual void ice_exception(const Ice::Exception&);
 
-private:
+protected:
+
+    void handleException(PyObject*);
 
     PyObject* _callback;
+};
 
-    void handleException(PyObject*);
+//
+// An asynchronous typed invocation with support for ice_sent.
+//
+class AsyncSentTypedInvocation : virtual public AsyncTypedInvocation, virtual public Ice::AMISentCallback
+{
+public:
+
+    AsyncSentTypedInvocation(const Ice::ObjectPrx&, const OperationPtr&);
+
+    virtual void ice_sent();
 };
 
 //
@@ -178,7 +190,7 @@ public:
     virtual void ice_response(bool, const pair<const Ice::Byte*, const Ice::Byte*>&);
     virtual void ice_exception(const Ice::Exception&);
 
-private:
+protected:
 
     string _op;
     PyObject* _callback;
@@ -187,6 +199,18 @@ private:
 };
 
 //
+// An asynchronous blobject invocation with support for ice_sent.
+//
+class AsyncSentBlobjectInvocation : virtual public AsyncBlobjectInvocation, virtual public Ice::AMISentCallback
+{
+public:
+
+    AsyncSentBlobjectInvocation(const Ice::ObjectPrx&);
+
+    virtual void ice_sent();
+};
+
+//
 // The base class for server-side upcalls.
 //
 class Upcall : virtual public IceUtil::Shared
@@ -242,7 +266,7 @@ private:
 };
 
 //
-// TypedServantWrapper uses the information in Operations to validate, marshal, and unmarshal
+// TypedServantWrapper uses the information in Operation to validate, marshal, and unmarshal
 // parameters and exceptions.
 //
 class TypedServantWrapper : public ServantWrapper
@@ -376,7 +400,7 @@ operationInvoke(OperationObject* self, PyObject* args)
     Ice::ObjectPrx prx = getProxy(pyProxy);
     assert(self->op);
 
-    InvocationPtr i = new SyncTypedInvocation(prx, OperationPtr::dynamicCast(*self->op));
+    InvocationPtr i = new SyncTypedInvocation(prx, *self->op);
     return i->invoke(opArgs);
 }
 
@@ -396,8 +420,22 @@ operationInvokeAsync(OperationObject* self, PyObject* args)
     Ice::ObjectPrx prx = getProxy(pyProxy);
     assert(self->op);
 
-    InvocationPtr i = new AsyncTypedInvocation(prx, OperationPtr::dynamicCast(*self->op));
-    return i->invoke(opArgs);
+    //
+    // If the callback implements an ice_sent method, we create a wrapper that derives
+    // from AMISentCallback.
+    //
+    assert(PyTuple_GET_SIZE(opArgs) > 0);
+    PyObject* callback = PyTuple_GET_ITEM(opArgs, 0);
+    if(PyObject_HasAttrString(callback, STRCAST("ice_sent")))
+    {
+        InvocationPtr i = new AsyncSentTypedInvocation(prx, *self->op);
+        return i->invoke(opArgs);
+    }
+    else
+    {
+        InvocationPtr i = new AsyncTypedInvocation(prx, *self->op);
+        return i->invoke(opArgs);
+    }
 }
 
 #ifdef WIN32
@@ -1121,7 +1159,7 @@ IcePy::AsyncTypedInvocation::~AsyncTypedInvocation()
 {
     AdoptThread adoptThread; // Ensure the current thread is able to call into Python.
 
-    Py_DECREF(_callback);
+    Py_XDECREF(_callback);
 }
 
 PyObject*
@@ -1144,6 +1182,7 @@ IcePy::AsyncTypedInvocation::invoke(PyObject* args)
         return 0;
     }
 
+    bool result = false;
     try
     {
         checkTwowayOnly(_prx);
@@ -1173,14 +1212,22 @@ IcePy::AsyncTypedInvocation::invoke(PyObject* args)
             }
 
             AllowThreads allowThreads; // Release Python's global interpreter lock during remote invocations.
-            _prx->ice_invoke_async(this, _op->name, _op->sendMode, pparams, ctx);
+            result = _prx->ice_invoke_async(this, _op->name, _op->sendMode, pparams, ctx);
         }
         else
         {
             AllowThreads allowThreads; // Release Python's global interpreter lock during remote invocations.
-            _prx->ice_invoke_async(this, _op->name, _op->sendMode, pparams);
+            result = _prx->ice_invoke_async(this, _op->name, _op->sendMode, pparams);
         }
     }
+    catch(const Ice::CommunicatorDestroyedException& ex)
+    {
+        //
+        // CommunicatorDestroyedException is the only exception that can propagate directly.
+        //
+        setPythonException(ex);
+        return 0;
+    }
     catch(const Ice::Exception& ex)
     {
         PyObjectHandle exh = convertException(ex);
@@ -1188,8 +1235,7 @@ IcePy::AsyncTypedInvocation::invoke(PyObject* args)
         handleException(exh.get());
     }
 
-    Py_INCREF(Py_None);
-    return Py_None;
+    PyRETURN_BOOL(result);
 }
 
 void
@@ -1293,6 +1339,38 @@ IcePy::AsyncTypedInvocation::handleException(PyObject* ex)
 }
 
 //
+// AsyncSentTypedInvocation
+//
+IcePy::AsyncSentTypedInvocation::AsyncSentTypedInvocation(const Ice::ObjectPrx& prx, const OperationPtr& op)
+    : Invocation(prx), TypedInvocation(prx, op), AsyncTypedInvocation(prx, op)
+{
+}
+
+void
+IcePy::AsyncSentTypedInvocation::ice_sent()
+{
+    AdoptThread adoptThread; // Ensure the current thread is able to call into Python.
+
+    PyObjectHandle method = PyObject_GetAttrString(_callback, STRCAST("ice_sent"));
+    if(!method.get())
+    {
+        ostringstream ostr;
+        ostr << "AMI callback object for operation `" << _op->name << "' does not define ice_sent()";
+        string str = ostr.str();
+        PyErr_Warn(PyExc_RuntimeWarning, const_cast<char*>(str.c_str()));
+    }
+    else
+    {
+        PyObjectHandle args = PyTuple_New(0);
+        PyObjectHandle tmp = PyObject_Call(method.get(), args.get(), 0);
+        if(PyErr_Occurred())
+        {
+            PyErr_Print();
+        }
+    }
+}
+
+//
 // SyncBlobjectInvocation
 //
 IcePy::SyncBlobjectInvocation::SyncBlobjectInvocation(const Ice::ObjectPrx& prx)
@@ -1418,7 +1496,7 @@ IcePy::AsyncBlobjectInvocation::~AsyncBlobjectInvocation()
 {
     AdoptThread adoptThread; // Ensure the current thread is able to call into Python.
 
-    Py_DECREF(_callback);
+    Py_XDECREF(_callback);
 }
 
 PyObject*
@@ -1460,12 +1538,13 @@ IcePy::AsyncBlobjectInvocation::invoke(PyObject* args)
         in.second = mem + sz;
     }
 
+    bool result = false;
     try
     {
         if(ctx == 0 || ctx == Py_None)
         {
             AllowThreads allowThreads; // Release Python's global interpreter lock during remote invocations.
-            _prx->ice_invoke_async(this, operation, sendMode, in);
+            result = _prx->ice_invoke_async(this, operation, sendMode, in);
         }
         else
         {
@@ -1476,9 +1555,17 @@ IcePy::AsyncBlobjectInvocation::invoke(PyObject* args)
             }
 
             AllowThreads allowThreads; // Release Python's global interpreter lock during remote invocations.
-            _prx->ice_invoke_async(this, operation, sendMode, in, context);
+            result = _prx->ice_invoke_async(this, operation, sendMode, in, context);
         }
     }
+    catch(const Ice::CommunicatorDestroyedException& ex)
+    {
+        //
+        // CommunicatorDestroyedException is the only exception that can propagate directly.
+        //
+        setPythonException(ex);
+        return 0;
+    }
     catch(const Ice::Exception& ex)
     {
         PyObjectHandle exh = convertException(ex);
@@ -1486,8 +1573,7 @@ IcePy::AsyncBlobjectInvocation::invoke(PyObject* args)
         handleException(exh.get());
     }
 
-    Py_INCREF(Py_None);
-    return Py_None;
+    PyRETURN_BOOL(result);
 }
 
 void
@@ -1605,6 +1691,38 @@ IcePy::AsyncBlobjectInvocation::handleException(PyObject* ex)
 }
 
 //
+// AsyncSentBlobjectInvocation
+//
+IcePy::AsyncSentBlobjectInvocation::AsyncSentBlobjectInvocation(const Ice::ObjectPrx& prx)
+    : Invocation(prx), AsyncBlobjectInvocation(prx)
+{
+}
+
+void
+IcePy::AsyncSentBlobjectInvocation::ice_sent()
+{
+    AdoptThread adoptThread; // Ensure the current thread is able to call into Python.
+
+    PyObjectHandle method = PyObject_GetAttrString(_callback, STRCAST("ice_sent"));
+    if(!method.get())
+    {
+        ostringstream ostr;
+        ostr << "AMI callback object for ice_invoke_async does not define ice_sent()";
+        string str = ostr.str();
+        PyErr_Warn(PyExc_RuntimeWarning, const_cast<char*>(str.c_str()));
+    }
+    else
+    {
+        PyObjectHandle args = PyTuple_New(0);
+        PyObjectHandle tmp = PyObject_Call(method.get(), args.get(), 0);
+        if(PyErr_Occurred())
+        {
+            PyErr_Print();
+        }
+    }
+}
+
+//
 // TypedUpcall
 //
 IcePy::TypedUpcall::TypedUpcall(const OperationPtr& op, const Ice::AMD_Array_Object_ice_invokePtr& callback,
@@ -2184,8 +2302,22 @@ IcePy::iceInvoke(const Ice::ObjectPrx& prx, PyObject* args)
 PyObject*
 IcePy::iceInvokeAsync(const Ice::ObjectPrx& prx, PyObject* args)
 {
-    InvocationPtr i = new AsyncBlobjectInvocation(prx);
-    return i->invoke(args);
+    //
+    // If the callback implements an ice_sent method, we create a wrapper that derives
+    // from AMISentCallback.
+    //
+    assert(PyTuple_GET_SIZE(args) > 0);
+    PyObject* callback = PyTuple_GET_ITEM(args, 0);
+    if(PyObject_HasAttrString(callback, STRCAST("ice_sent")))
+    {
+        InvocationPtr i = new AsyncSentBlobjectInvocation(prx);
+        return i->invoke(args);
+    }
+    else
+    {
+        InvocationPtr i = new AsyncBlobjectInvocation(prx);
+        return i->invoke(args);
+    }
 }
 
 //
@@ -2253,6 +2385,7 @@ IcePy::TypedServantWrapper::ice_invoke_async(const Ice::AMD_Array_Object_ice_inv
                                                           const_cast<char*>(attrName.c_str()));
                 if(!h.get())
                 {
+                    PyErr_Clear();
                     Ice::OperationNotExistException ex(__FILE__, __LINE__);
                     ex.id = current.id;
                     ex.facet = current.facet;
diff --git a/py/modules/IcePy/Proxy.cpp b/py/modules/IcePy/Proxy.cpp
index 60c1d993fd6..7b84a8c3f03 100644
--- a/py/modules/IcePy/Proxy.cpp
+++ b/py/modules/IcePy/Proxy.cpp
@@ -1445,6 +1445,158 @@ proxyIceGetCachedConnection(ProxyObject* self)
 extern "C"
 #endif
 static PyObject*
+proxyIceFlushBatchRequests(ProxyObject* self)
+{
+    assert(self->proxy);
+
+    try
+    {
+        (*self->proxy)->ice_flushBatchRequests();
+    }
+    catch(const Ice::Exception& ex)
+    {
+        setPythonException(ex);
+        return 0;
+    }
+
+    Py_INCREF(Py_None);
+    return Py_None;
+}
+
+namespace IcePy
+{
+class AMI_Object_ice_flushBatchRequestsI : public Ice::AMI_Object_ice_flushBatchRequests
+{
+public:
+
+    AMI_Object_ice_flushBatchRequestsI(PyObject*);
+    ~AMI_Object_ice_flushBatchRequestsI();
+
+    virtual void ice_exception(const Ice::Exception&);
+
+protected:
+
+    PyObject* _callback;
+};
+
+class AMI_Object_ice_flushBatchRequestsSentI : public AMI_Object_ice_flushBatchRequestsI, public Ice::AMISentCallback
+{
+public:
+
+    AMI_Object_ice_flushBatchRequestsSentI(PyObject*);
+
+    virtual void ice_sent();
+};
+
+AMI_Object_ice_flushBatchRequestsI::AMI_Object_ice_flushBatchRequestsI(PyObject* callback) :
+    _callback(callback)
+{
+    Py_INCREF(callback);
+}
+
+AMI_Object_ice_flushBatchRequestsI::~AMI_Object_ice_flushBatchRequestsI()
+{
+    AdoptThread adoptThread; // Ensure the current thread is able to call into Python.
+
+    Py_DECREF(_callback);
+}
+
+void
+AMI_Object_ice_flushBatchRequestsI::ice_exception(const Ice::Exception& ex)
+{
+    AdoptThread adoptThread; // Ensure the current thread is able to call into Python.
+
+    PyObjectHandle method = PyObject_GetAttrString(_callback, STRCAST("ice_exception"));
+    if(!method.get())
+    {
+        ostringstream ostr;
+        ostr << "AMI callback object for ice_flushBatchRequests_async does not define ice_exception()";
+        string str = ostr.str();
+        PyErr_Warn(PyExc_RuntimeWarning, const_cast<char*>(str.c_str()));
+    }
+    else
+    {
+        PyObjectHandle exh = convertException(ex);
+        assert(exh.get());
+
+        PyObjectHandle args = Py_BuildValue(STRCAST("(O)"), exh.get());
+        PyObjectHandle tmp = PyObject_Call(method.get(), args.get(), 0);
+        if(PyErr_Occurred())
+        {
+            PyErr_Print();
+        }
+    }
+}
+
+AMI_Object_ice_flushBatchRequestsSentI::AMI_Object_ice_flushBatchRequestsSentI(PyObject* callback) :
+    AMI_Object_ice_flushBatchRequestsI(callback)
+{
+}
+
+void
+AMI_Object_ice_flushBatchRequestsSentI::ice_sent()
+{
+    AdoptThread adoptThread; // Ensure the current thread is able to call into Python.
+
+    PyObjectHandle method = PyObject_GetAttrString(_callback, STRCAST("ice_sent"));
+    if(!method.get())
+    {
+        ostringstream ostr;
+        ostr << "AMI callback object for ice_flushBatchRequests_async does not define ice_sent()";
+        string str = ostr.str();
+        PyErr_Warn(PyExc_RuntimeWarning, const_cast<char*>(str.c_str()));
+    }
+    else
+    {
+        PyObjectHandle args = PyTuple_New(0);
+        PyObjectHandle tmp = PyObject_Call(method.get(), args.get(), 0);
+        if(PyErr_Occurred())
+        {
+            PyErr_Print();
+        }
+    }
+}
+}
+
+#ifdef WIN32
+extern "C"
+#endif
+static PyObject*
+proxyIceFlushBatchRequestsAsync(ProxyObject* self, PyObject* args)
+{
+    assert(self->proxy);
+
+    PyObject* cb;
+    if(!PyArg_ParseTuple(args, STRCAST("O"), &cb))
+    {
+        return 0;
+    }
+
+    bool result = false;
+    try
+    {
+        if(PyObject_HasAttrString(cb, STRCAST("ice_sent")))
+        {
+            result = (*self->proxy)->ice_flushBatchRequests_async(new AMI_Object_ice_flushBatchRequestsSentI(cb));
+        }
+        else
+        {
+            result = (*self->proxy)->ice_flushBatchRequests_async(new AMI_Object_ice_flushBatchRequestsI(cb));
+        }
+    }
+    catch(const Ice::Exception& ex)
+    {
+        setPythonException(ex);
+        return 0;
+    }
+
+    PyRETURN_BOOL(result);
+}
+
+#ifdef WIN32
+extern "C"
+#endif
+static PyObject*
 proxyIceInvoke(ProxyObject* self, PyObject* args)
 {
     return iceInvoke(*self->proxy, args);
@@ -1824,10 +1976,14 @@ static PyMethodDef ProxyMethods[] =
         PyDoc_STR(STRCAST("ice_getConnection() -> Ice.Connection")) },
     { STRCAST("ice_getCachedConnection"), reinterpret_cast<PyCFunction>(proxyIceGetCachedConnection), METH_NOARGS,
         PyDoc_STR(STRCAST("ice_getCachedConnection() -> Ice.Connection")) },
+    { STRCAST("ice_flushBatchRequests"), reinterpret_cast<PyCFunction>(proxyIceFlushBatchRequests), METH_VARARGS,
+        PyDoc_STR(STRCAST("ice_flushBatchRequests() -> void")) },
+    { STRCAST("ice_flushBatchRequests_async"), reinterpret_cast<PyCFunction>(proxyIceFlushBatchRequestsAsync),
+        METH_VARARGS, PyDoc_STR(STRCAST("ice_flushBatchRequests_async(cb) -> bool")) },
     { STRCAST("ice_invoke"), reinterpret_cast<PyCFunction>(proxyIceInvoke), METH_VARARGS,
         PyDoc_STR(STRCAST("ice_invoke(operation, mode, inParams) -> bool, outParams")) },
     { STRCAST("ice_invoke_async"), reinterpret_cast<PyCFunction>(proxyIceInvokeAsync), METH_VARARGS,
-        PyDoc_STR(STRCAST("ice_invoke_async(cb, operation, mode, inParams) -> void")) },
+        PyDoc_STR(STRCAST("ice_invoke_async(cb, operation, mode, inParams) -> bool")) },
     { STRCAST("ice_checkedCast"), reinterpret_cast<PyCFunction>(proxyIceCheckedCast), METH_VARARGS | METH_CLASS,
         PyDoc_STR(STRCAST("ice_checkedCast(proxy, id[, facetOrCtx[, ctx]]) -> proxy")) },
     { STRCAST("ice_uncheckedCast"), reinterpret_cast<PyCFunction>(proxyIceUncheckedCast), METH_VARARGS | METH_CLASS,
diff --git a/py/modules/IcePy/Util.h b/py/modules/IcePy/Util.h
index 5e80348796d..c05aacc41b3 100644
--- a/py/modules/IcePy/Util.h
+++ b/py/modules/IcePy/Util.h
@@ -26,6 +26,8 @@
 #define PyRETURN_FALSE return Py_INCREF(getFalse()), getFalse()
 #define PyRETURN_TRUE return Py_INCREF(getTrue()), getTrue()
 
+#define PyRETURN_BOOL(b) if(b) PyRETURN_TRUE; else PyRETURN_FALSE
+
 namespace IcePy
 {
 
diff --git a/py/test/Ice/operations/TwowaysAMI.py b/py/test/Ice/operations/TwowaysAMI.py
index 3987ab93e8c..022e48b6079 100644
--- a/py/test/Ice/operations/TwowaysAMI.py
+++ b/py/test/Ice/operations/TwowaysAMI.py
@@ -527,17 +527,17 @@ def twowaysAMI(communicator, p):
     indirect = Test.MyClassPrx.uncheckedCast(p.ice_adapterId("dummy"))
     cb = AMI_MyClass_opVoidExI()
     try:
-        indirect.opVoid_async(cb)
+        test(not indirect.opVoid_async(cb))
     except Ice.Exception:
         test(False)
     test(cb.check())
 
-    # Check that a call to a void operation raises NoEndpointException
+    # Check that a call to a twoway operation raises NoEndpointException
     # in the ice_exception() callback instead of at the point of call.
     indirect = Test.MyClassPrx.uncheckedCast(p.ice_adapterId("dummy"))
     cb = AMI_MyClass_opByteExI()
     try:
-        indirect.opByte_async(cb, 0, 0)
+        test(not indirect.opByte_async(cb, 0, 0))
     except Ice.Exception:
         test(False)
     test(cb.check())
@@ -552,6 +552,22 @@ def twowaysAMI(communicator, p):
     p.opVoid_async(cb)
     test(cb.check())
 
+    # Check that CommunicatorDestroyedException is raised directly.
+    initData = Ice.InitializationData()
+    initData.properties = communicator.getProperties().clone()
+    ic = Ice.initialize(initData)
+    obj = ic.stringToProxy(p.ice_toString())
+    p2 = Test.MyClassPrx.checkedCast(obj)
+
+    ic.destroy()
+
+    cb = AMI_MyClass_opVoidI()
+    try:
+        test(not p2.opVoid_async(cb))
+        test(False)
+    except Ice.CommunicatorDestroyedException:
+        pass # Expected.
+
     #
     # opByte
     #
@@ -904,8 +920,3 @@ def twowaysAMI(communicator, p):
         test(cb.check())
 
         ic.destroy()
-        
-
-
-
-