feat(server): Implement the Cancel services

src/server/ua_server_async.c

@@ -52,13 +52,13 @@ UA_AsyncManager_sendAsyncResponse(UA_AsyncManager *am, UA_Server *server,
     UA_StatusCode res =
         sendResponse(server, session, channel, ar->requestId,
                      (UA_Response*)&ar->response, &UA_TYPES[UA_TYPES_CALLRESPONSE]);
-    UA_AsyncManager_removeAsyncResponse(&server->asyncManager, ar);
     if(res != UA_STATUSCODE_GOOD) {
         UA_LOG_WARNING_SESSION(&server->config.logger, session,
                                "Async Response for Req# %" PRIu32 " failed "
                                "with StatusCode %s", ar->requestId,
                                UA_StatusCode_name(res));
     }
+    UA_AsyncManager_removeAsyncResponse(&server->asyncManager, ar);
 }
 
 /* Integrate operation result in the AsyncResponse and send out the response if
@@ -429,4 +429,47 @@ UA_Server_processServiceOperationsAsync(UA_Server *server, UA_Session *session,
     return UA_STATUSCODE_GOOD;
 }
 
+UA_UInt32
+UA_AsyncManager_cancel(UA_Server *server, UA_Session *session, UA_UInt32 requestHandle) {
+    UA_LOCK(&server->serviceMutex);
+    UA_AsyncManager *am = &server->asyncManager;
+
+    /* Loop over the queue of dispatched ops */
+    UA_AsyncOperation *op = NULL, *op_tmp = NULL;
+    TAILQ_FOREACH_SAFE(op, &am->dispatchedQueue, pointers, op_tmp) {
+        if(op->parent->requestHandle != requestHandle ||
+           !UA_NodeId_equal(&session->sessionId, &op->parent->sessionId))
+            continue;
+
+        /* Set status and put it into the result queue */
+        op->response.statusCode = UA_STATUSCODE_BADREQUESTCANCELLEDBYCLIENT;
+        TAILQ_REMOVE(&am->dispatchedQueue, op, pointers);
+        TAILQ_INSERT_TAIL(&am->resultQueue, op, pointers);
+
+        /* Also set the status of the overall response */
+        op->parent->response.callResponse.responseHeader.
+            serviceResult = UA_STATUSCODE_BADREQUESTCANCELLEDBYCLIENT;
+    }
+
+    /* Idem for waiting ops */
+    TAILQ_FOREACH_SAFE(op, &am->newQueue, pointers, op_tmp) {
+        if(op->parent->requestHandle != requestHandle ||
+           !UA_NodeId_equal(&session->sessionId, &op->parent->sessionId))
+            continue;
+
+        /* Mark as timed out and put it into the result queue */
+        op->response.statusCode = UA_STATUSCODE_BADREQUESTCANCELLEDBYCLIENT;
+        TAILQ_REMOVE(&am->newQueue, op, pointers);
+        TAILQ_INSERT_TAIL(&am->resultQueue, op, pointers);
+
+        op->parent->response.callResponse.responseHeader.
+            serviceResult = UA_STATUSCODE_BADREQUESTCANCELLEDBYCLIENT;
+    }
+
+    UA_UNLOCK(&server->serviceMutex);
+
+    /* Process messages that have all ops completed */
+    return processAsyncResults(server);
+}
+
 #endif

src/server/ua_server_async.h

@@ -93,6 +93,12 @@ UA_AsyncManager_createAsyncOp(UA_AsyncManager *am, UA_Server *server,
                               UA_AsyncResponse *ar, size_t opIndex,
                               const UA_CallMethodRequest *opRequest);
 
+/* Send out the response with status set. Also removes all outstanding
+ * operations from the dispatch queue. The queuelock needs to be taken before
+ * calling _cancel. */
+UA_UInt32
+UA_AsyncManager_cancel(UA_Server *server, UA_Session *session, UA_UInt32 requestHandle);
+
 typedef void (*UA_AsyncServiceOperation)(UA_Server *server, UA_Session *session,
                                          UA_UInt32 requestId, UA_UInt32 requestHandle,
                                          size_t opIndex, const void *requestOperation,

src/server/ua_server_binary.c

@@ -280,6 +280,11 @@ getServicePointers(UA_UInt32 requestTypeId, const UA_DataType **requestType,
         *requestType = &UA_TYPES[UA_TYPES_CLOSESESSIONREQUEST];
         *responseType = &UA_TYPES[UA_TYPES_CLOSESESSIONRESPONSE];
         break;
+    case UA_NS0ID_CANCELREQUEST_ENCODING_DEFAULTBINARY:
+        *service = (UA_Service)Service_Cancel;
+        *requestType = &UA_TYPES[UA_TYPES_CANCELREQUEST];
+        *responseType = &UA_TYPES[UA_TYPES_CANCELRESPONSE];
+        break;
     case UA_NS0ID_READREQUEST_ENCODING_DEFAULTBINARY:
         *service = (UA_Service)Service_Read;
         *requestType = &UA_TYPES[UA_TYPES_READREQUEST];

src/server/ua_services.h

@@ -172,7 +172,9 @@ void Service_CloseSession(UA_Server *server, UA_SecureChannel *channel,
  * ^^^^^^^^^^^^^^
  * Used to cancel outstanding Service requests. Successfully cancelled service
  * requests shall respond with Bad_RequestCancelledByClient. */
-/* Not Implemented */
+void Service_Cancel(UA_Server *server, UA_Session *session,
+                    const UA_CancelRequest *request,
+                    UA_CancelResponse *response);
 
 /**
  * NodeManagement Service Set

src/server/ua_services_session.c

@@ -940,3 +940,44 @@ Service_CloseSession(UA_Server *server, UA_SecureChannel *channel,
         UA_Server_removeSessionByToken(server, &session->header.authenticationToken,
                                        UA_SHUTDOWNREASON_CLOSE);
 }
+
+void Service_Cancel(UA_Server *server, UA_Session *session,
+                    const UA_CancelRequest *request,
+                    UA_CancelResponse *response) {
+    /* If multithreading is disabled, then there are no async services. If all
+     * services are answered "right away", then there are no services that can
+     * be cancelled. */
+#if UA_MULTITHREADING >= 100
+    response->cancelCount = UA_AsyncManager_cancel(server, session, request->requestHandle);
+#endif
+
+    /* Publish requests for Subscriptions are stored separately */
+#ifdef UA_ENABLE_SUBSCRIPTIONS
+    UA_PublishResponseEntry *pre, *pre_tmp;
+    UA_PublishResponseEntry *prev = NULL;
+    SIMPLEQ_FOREACH_SAFE(pre, &session->responseQueue, listEntry, pre_tmp) {
+        /* Skip entry and set as the previous entry that is kept in the list */
+        if(pre->response.responseHeader.requestHandle != request->requestHandle) {
+            prev = pre;
+            continue;
+        }
+
+        /* Dequeue */
+        if(prev)
+            SIMPLEQ_REMOVE_AFTER(&session->responseQueue, prev, listEntry);
+        else
+            SIMPLEQ_REMOVE_HEAD(&session->responseQueue, listEntry);
+        session->responseQueueSize--;
+
+        /* Send response and clean up */
+        response->responseHeader.serviceResult = UA_STATUSCODE_BADREQUESTCANCELLEDBYCLIENT;
+        sendResponse(server, session, session->header.channel, pre->requestId,
+                     (UA_Response *)response, &UA_TYPES[UA_TYPES_PUBLISHRESPONSE]);
+        UA_PublishResponse_clear(&pre->response);
+        UA_free(pre);
+
+        /* Increase the CancelCount */
+        response->cancelCount++;
+    }
+#endif
+}

tools/schema/datatypes_minimal.txt

@@ -117,4 +117,6 @@ ServerState
 ServerDiagnosticsSummaryDataType
 RedundancySupport
 KeyValuePair
-NamingRuleType
+NamingRuleType
+CancelRequest
+CancelResponse