[Core] Keep processing commands if submit future completes exceptionally

Author: alexrobin

sensorhub-core/src/main/java/org/sensorhub/api/command/IStreamingControlInterface.java

@@ -107,9 +107,10 @@ public default DataEncoding getCommandEncoding()
      * are then sent via the dedicated event channel, using the same task ID,
      * to provide status updates to the caller.
      * </p><p>
-     * The future should only complete exceptionally if there is an unexpected error.
-     * All errors associated to the processing of the command by the receiver system
-     * should be reported via a status object instead.
+     * The future should only complete exceptionally if there is an unexpected error
+     * or internal error that should not be reported to the caller. Otherwise, all
+     * errors associated to the processing of the command by the receiver system
+     * should be reported in the status object instead.
      * </p>
      * @param command Command data (with ID set)
      * @return A future that will be completed normally when the system is ready to

sensorhub-core/src/main/java/org/sensorhub/impl/system/SystemDriverTransactionHandler.java

@@ -365,7 +365,8 @@ protected void connectControlInput(IStreamingControlInterface controlInput, Comm
     {
         csHandler.connectCommandReceiver(new Subscriber<CommandEvent>() {
             Subscription sub;
-            static final String ERROR_MESSAGE = "Error sending command to {}. Canceling command receiver subscription";
+            static final String ERROR_MESSAGE = "Error dispatching command to {}. ";
+            static final String DISPATCH_STOP_MESSAGE = "No more commands will be processed.";
 
             @Override
             public void onSubscribe(Subscription sub)
@@ -388,14 +389,16 @@ public void onNext(CommandEvent event)
                             })
                             .exceptionally(e -> {
                                 DefaultSystemRegistry.log.error(ERROR_MESSAGE, csHandler.csInfo.getFullName(), e);
-                                sub.cancel();
-                                csHandler.sendStatus(event.getCorrelationID(), CommandStatus.failed(event.getCommand().getID(), "Internal error processing command"));
+                                csHandler.sendStatus(event.getCorrelationID(),
+                                    CommandStatus.failed(event.getCommand().getID(), "Internal error processing command"));
+                                sub.request(1);
                                 return null; // return type is Void
                             });
                     }
                     catch (Exception e)
                     {
-                        DefaultSystemRegistry.log.error(ERROR_MESSAGE, csHandler.csInfo.getFullName(), e);
+                        DefaultSystemRegistry.log.error(ERROR_MESSAGE + DISPATCH_STOP_MESSAGE,
+                            csHandler.csInfo.getFullName(), e);
                         sub.cancel();
                     }
                 });
@@ -404,8 +407,8 @@ public void onNext(CommandEvent event)
             @Override
             public void onError(Throwable e)
             {
-                DefaultSystemRegistry.log.error("Error dispatching commands to {}. "
-                    + "No more commands will be processed.", csHandler.csInfo.getFullName(), e);
+                DefaultSystemRegistry.log.error(ERROR_MESSAGE + DISPATCH_STOP_MESSAGE,
+                    csHandler.csInfo.getFullName(), e);
             }
 
             @Override