MessageQ_put returns MessageQ_E_SHUTDOWN when transport fails

[ipc/ipcdev.git] / linux / src / api / MessageQ.c

diff --git a/linux/src/api/MessageQ.c b/linux/src/api/MessageQ.c
index b229e1b26193c28038929feeb2068fae1ec0ebdd..13ab27d5be1f4e548e87a20b237d1b43aac323ce 100644 (file)
--- a/linux/src/api/MessageQ.c
+++ b/linux/src/api/MessageQ.c
@@ -860,7 +860,9 @@ Int MessageQ_put(MessageQ_QueueId queueId, MessageQ_Msg msg)
             case INetworkTransport_TypeId:
                 netTrans = INetworkTransport_downCast(baseTrans);
                 delivered = INetworkTransport_put(netTrans, (Ptr)msg);
-                status = (delivered ? MessageQ_S_SUCCESS : MessageQ_E_FAIL);
+                status = (delivered ? MessageQ_S_SUCCESS :
+                          (errno == ESHUTDOWN ? MessageQ_E_SHUTDOWN :
+                           MessageQ_E_FAIL));
                 break;
 
             default:
@@ -886,7 +888,8 @@ Int MessageQ_put(MessageQ_QueueId queueId, MessageQ_Msg msg)
 
         msgTrans = MessageQ_module->transports[clusterId][priority];
         delivered = IMessageQTransport_put(msgTrans, (Ptr)msg);
-        status = (delivered ? MessageQ_S_SUCCESS : MessageQ_E_FAIL);
+        status = (delivered ? MessageQ_S_SUCCESS :
+                  (errno == ESHUTDOWN ? MessageQ_E_SHUTDOWN : MessageQ_E_FAIL));
     }
 
 done:
@@ -957,7 +960,7 @@ Int MessageQ_get(MessageQ_Handle handle, MessageQ_Msg *msg, UInt timeout)
     }
 
     if (obj->unblocked) {
-        return MessageQ_E_UNBLOCKED;
+        return obj->unblocked;
     }
 
     pthread_mutex_lock(&MessageQ_module->gate);
@@ -1073,7 +1076,16 @@ Void MessageQ_unblock(MessageQ_Handle handle)
 {
     MessageQ_Object *obj = (MessageQ_Object *)handle;
 
-    obj->unblocked = TRUE;
+    obj->unblocked = MessageQ_E_UNBLOCKED;
+    sem_post(&obj->synchronizer);
+}
+
+/* Unblocks a MessageQ that's been shutdown due to transport failure */
+Void MessageQ_shutdown(MessageQ_Handle handle)
+{
+    MessageQ_Object *obj = (MessageQ_Object *)handle;
+
+    obj->unblocked = MessageQ_E_SHUTDOWN;
     sem_post(&obj->synchronizer);
 }
 
@@ -1097,6 +1109,24 @@ MessageQ_QueueId MessageQ_getQueueId(MessageQ_Handle handle)
     return queueId;
 }
 
+/* Returns the local handle associated with queueId. */
+MessageQ_Handle MessageQ_getLocalHandle(MessageQ_QueueId queueId)
+{
+    MessageQ_Object *obj;
+    MessageQ_QueueIndex queueIndex;
+    UInt16 procId;
+
+    procId = MessageQ_getProcId(queueId);
+    if (procId != MultiProc_self()) {
+        return NULL;
+    }
+
+    queueIndex = MessageQ_getQueueIndex(queueId);
+    obj = (MessageQ_Object *)MessageQ_module->queues[queueIndex];
+
+    return (MessageQ_Handle)obj;
+}
+
 /* Sets the tracing of a message */
 Void MessageQ_setMsgTrace(MessageQ_Msg msg, Bool traceFlag)
 {