Communication using Commands

The csw-command library provides support for command based communication between components.

This section describes how to communicate with any other component using commands. To check how to manage commands received, please visit Component Handlers and Managing Command State.

Dependencies

sbt

libraryDependencies += "com.github.tmtsoftware.csw" %% "csw-command" % "5.0.1"

Command-based Communication Between Components

A component can send Commands to other components. The commands can be sent as one of the following three types of messages:

• submit - A command is sent using a submit when the result of completion is desired.
• oneway - A command is sent as oneway when the result of completion is not desired.
• validate - A command is sent, but it is only validated with no actions started.

Submit comes in two versions: submit and submitAndWait. The only difference is in how long-running commands are treated. If a command starts long-running actions, submit returns a Started response immediately while submitAndWait will wait for the long-running actions to complete before returning a final response. The submitAndWait is a composite of submit and queryFinal.

A submitAndWait is the most convenient way of sending a command from one component to another. When received, a submitAndWait command is validated and if accepted, the actions associated with the command are executed or started. When the submit response that is part of the submitAndWait is received, submitAndWait immediately does a queryFinal. For simple scenarios, this is the right command and will be the most commonly used.

A submit can be used when the sender of a command needs to do additional work before long-running actions are completed. For instance, send another command to execute in parallel. If commands are short, submit and submitAndWait behave the same way. When the actions started by submit are long-running, the caller can wait for the actions to complete if needed using the queryFinal call.

A oneway is primarily meant to be used between an Assembly and an HCD when no completion information is desired. It is also useful when tracking completion using a Matcher and current state values (see below) or events.

A validate message is used to ask a destination component to validate a command and determine if the command can be executed. It does not execute the command and only returns the result of validation. In some scenarios, it may be useful to test to see if a command can be executed prior to trying to execute the command.

Component Locking

If a component is locked, any Command Service command will have a response of Locked. The corresponding handler will not be called until the component is unlocked. See Creating a Component for more information on locking.

Note that the code in the receiving component’s handler does not need to return Locked. If the component has been Locked, the component’s Supervisor returns the Locked response to the caller and the handler is not called.

Command Validation

Command validations occur in two scenarios. One is using the validate message as described above. For example, validate could be used by an Assembly when it needs to send multiple commands to different HCDs and it wants to first check that all the HCDs can execute their commands before sending a command to any of the HCDs.
The second scenario is that it is always the first step in processing a command with submit, submitAndWait, or oneway.

If the receiving component is not locked, the component’s supervisor calls the validateCommand handler of the Top Level Actor. The developer code evaluates and returns a ValidateCommandResponse as shown in the following table.

ValidateCommandResponse	Description
Accepted	The command is valid and can be executed by the component.
Invalid	The command is not valid and cannot be executed. The response includes a reason in the form of a CommandIssue
Locked	The component is locked by some other command sender. The validation could not occur.

The Submit Message

A submit message is sent with its command to a component destination. A SubmitResponse is returned to the caller when the submit message is used. If the validateCommand handler returns Accepted, the framework calls the onSubmit handler of the Top Level Actor. The onSubmit handler always returns a SubmitResponse.

Immediate Completion Scenario (submit)

If the actions of the submit command take a very short time to complete, they may be completed by the onSubmit handler. This is called immediate completion. The time for the actions to complete should be less than 1 second. (Note: The framework will timeout if the destination does not return a response within 1 second.) In this scenario with onSubmit, the values of SubmitResponse can be Completed or Error. Error is returned when the actions could not be accomplished. This is different than Invalid, which indicates that the command could not be validated.

The immediate completion behavior is similar to a remote procedure call although the execution is entirely asynchronous. If the actions are successful, the Completed SubmitResponse is returned. If there is a result, the Completed is returned with a parameter set of Result type that can be inspected by the sender.

Long Running Actions Scenario (submit)

When actions take longer than 1 second, onSubmit should start the actions and return the Started SubmitResponse. The Started response indicates to the framework that long-running actions have been started.

Once the long-running actions have started, the receiving component code must notify the framework when the actions are completed. This is done be updating through the Command Response Manager.

In addition to the values returned for immediate completion, long running actions can return Cancelled. If the component supports a separate command to stop a long-running command, the stopped command should return Cancelled when successfully cancelled. The command that cancels the long running command should return Completed.

The SubmitAndWait Message

A submitAndWait message is sent with its command to a component destination. A SubmitResponse is returned to the caller when the submitAndWait message is used. If the validateCommand handler returns Accepted, the framework calls the onSubmit handler of the Top Level Actor. The onSubmit handler always returns a SubmitResponse.

Immediate Completion Scenario (submitAndWait)

If the actions of the submitAndWait command take a very short time to complete, submitAndWait behaves exactly like submit as described above.

The immediate completion behavior is similar to a remote procedure call although the execution is entirely asynchronous. If the actions are successful, the Completed SubmitResponse is returned. If there is a result, the Completed is returned with a parameter set of Result type that can be inspected by the sender.

Long Running Actions Scenario (submitAndWait)

When actions take longer than 1 second, onSubmit should start the actions and return the Started SubmitResponse. The Started response indicates to the framework that long-running actions have been started. In this case, submitAndWait immediately executes a queryFinal command in order to wait for the final response.

As with the submit description, the receiving component code must notify the framework when the actions are completed. This is done be updating the Command Response Manager.

SubmitResponse Summary

The following table summarizes all the possible values for SubmitResponse.

SubmitResponse	Description
Invalid	The command is not valid and cannot be executed. The response includes a reason in the form of a CommandIssue. onSubmit is not executed.
Completed(Result)	This response is returned when the actions associated with a command are completed. A result may or may not be returned.
Started	Returned by onSubmit when long-running actions have been started.
Error	Error is returned the the actions started by a command do not complete properly. A message is returned explaining the error.
Cancelled	The actions associated with a long-running command have been cancelled.
Locked	The component is locked by some other command sender. The validation could not occur.

The Oneway Message

The other option for sending a command to a component destination is the oneway message. The central difference between submit and oneway is that oneway does not track or allow reporting of completion information. It supports a fire and forget type of communication approach.

A OnewayResponse is returned to the caller when the oneway message is used. Oneway does validate the command. If the component is not locked, the validateCommand handler is called.
If the validateCommand handler returns Accepted, the framework calls the onOneway handler of the Top Level Actor. However, the onOneway handler does not return a value. The sender of the oneway message receives the result of the validation or the Locked indication.

The following table summarizes all the possible values for OnewayResponse.

OnewayResponse	Description
Invalid	The command is not valid and cannot be executed. The response includes a reason in the form of a CommandIssue. onOneway is not executed.
Accepted	Returned when validation succeeds and the command was passed to the onOneway handler.
Locked	The component is locked by some other command sender. The validation could not occur.

Oneway is available as a higher performance option when an Assembly needs to send commands to an HCD but doesn’t really care about completion such as the case when demands are being sent to a motor. Validation is still present to ensure the HCD supports the standalone operation requirement that allows the HCD to check that its input values are not invalid and it is not getting out of range values.

Oneway can be used with a matcher. The matcher can use CurrentState or even events from the Event Service to determine completion. This can be more complicated than submit, but may be useful in some scenarios.

CommandService

A helper/wrapper is provided called CommandService that provides a convenient way to use the Command Service with a component from the Location Service.

A CommandService instance is created using an AkkaLocation of the receiving component, discovered from the Location Service. This CommandService instance has methods for communicating with the component. A new CommandService is created for each component for which commands are to be sent.

This example shows creating a Command Service using Akka Location of the receiving component, discovered from the Location Service. Actor System is created here using Actor System Context ctx and is passed as implicit parameter to make method of CommandServiceFactory. In Java though the actor is passed explicitly as shown in the example.

Scala/resolve hcd/create command service

copysourceclass SampleAssemblyHandlers(ctx: ActorContext[TopLevelActorMessage], cswCtx: CswContext) extends ComponentHandlers(ctx, cswCtx) {
  private implicit val system: ActorSystem[Nothing] = ctx.system
  override def onLocationTrackingEvent(trackingEvent: TrackingEvent): Unit = {
    log.debug(s"onLocationTrackingEvent called: $trackingEvent")
    trackingEvent match {
      case LocationUpdated(location) =>
        hcdLocation = location.asInstanceOf[AkkaLocation]
        hcdCS = Some(CommandServiceFactory.make(location))
        onSetup(Id(), Setup(prefix, shortCommand, None))
      case LocationRemoved(connection) =>
        if (connection == hcdConnection) {
          hcdCS = None
        }
    }
  }

Java/resolve hcd/create command service

copysourceprivate final ActorSystem<Void> system;
JSampleAssemblyHandlers(ActorContext<TopLevelActorMessage> ctx, JCswContext cswCtx) {
  super(ctx, cswCtx);
  this.cswCtx = cswCtx;
  system = ctx.getSystem();
@Override
public void onLocationTrackingEvent(TrackingEvent trackingEvent) {
  log.debug(() -> "onLocationTrackingEvent called: " + trackingEvent.toString());
  if (trackingEvent instanceof LocationUpdated) {
    Location location = ((LocationUpdated) trackingEvent).location();
    hcdLocation = (AkkaLocation) location;
    cswCtx.eventService().defaultPublisher().publish(new SystemEvent(prefix, new EventName("receivedHcdLocation")));
    hcdCS = Optional.of(CommandServiceFactory.jMake(location, system));
    onSetup(Id.apply(), new Setup(prefix, shortCommand, Optional.empty()));
  } else if (trackingEvent instanceof LocationRemoved) {
    Connection connection = trackingEvent.connection();
    if (connection == hcdConnection) {
      hcdCS = Optional.empty();
    }
  }
}

The API can be exercised as follows for different scenarios of command-based communication:

submit

Sending a submit message with a command returns a SubmitResponse as a Future. The Future returned by submit will be the final response in case of short-running command and may be a positive completion (Completed) or a negative completion (Invalid, Error, Cancelled, Locked). The response may also be the Started response in the case of long-running actions. In the case of a long-running command the Started response can be used with either query or queryFinal to obtain the final response.

This example shows a command using submit that returns Started followed by a queryFinal to get the final response.

Scala/submit w/queryFinal

copysourceprivate def sleepHCD(runId: Id, setup: Setup, sleepTime: Long): Unit =
  hcdCS match {
    case Some(cs) =>
      val s = Setup(prefix, hcdSleep, None).add(setSleepTime(sleepTime))
      cs.submit(s).foreach {
        case started: Started =>
          // Can insert extra code during execution here
          cs.queryFinal(started.runId).foreach(sr => commandResponseManager.updateCommand(sr.withRunId(runId)))
        case other =>
          commandResponseManager.updateCommand(other.withRunId(runId))
      }
    case None =>
      commandResponseManager.updateCommand(
        Error(runId, s"A needed HCD is not available: ${hcdConnection.componentId} for $prefix")
      )
  }

Java/submit w/queryFinal

copysourceprivate void sleepHCD(Id runId, Setup setup, Long sleepTime) {
  if (hcdCS.isPresent()) {
    ICommandService cs = hcdCS.get();
    Setup s = new Setup(prefix, hcdSleep, Optional.empty()).add(setSleepTime(sleepTime));
    cs.submit(s).thenAccept(submitResponse -> {
      if (submitResponse instanceof Started) {
        Started started = (Started) submitResponse;
        // Can insert extra code during execution here
        cs.queryFinal(started.runId(), timeout).thenAccept(sr -> cswCtx.commandResponseManager().updateCommand(sr.withRunId(runId)));
      } else {
        cswCtx.commandResponseManager().updateCommand(submitResponse.withRunId(runId));
      }
    });
  } else {
    cswCtx.commandResponseManager().updateCommand(
        new CommandResponse.Error(runId, "A needed HCD is not available: " + hcdConnection.componentId() + " for " + prefix)
    );
  }
}

If using submit and the validation fails in the destination component, the Invalid response is returned. Sometimes it is possible to handle an Invalid response locally, but in most cases it must simply be returned to the caller. The following example shows how to process individual responses from a submit:

Scala/submit w/invalid response

copysourceval invalidSetup    = Setup(prefix, invalidCmd, obsId)
val invalidCommandF = assemblyCmdService.submitAndWait(invalidSetup)
val resultF: Future[Unit] = async {
  await(invalidCommandF) match {
    case Completed(_, _) =>
    // Do Completed thing
    case Invalid(_, _) =>
    // issue shouldBe a[Invalid]
    case other =>
      // Unexpected result
      log.error(s"Some other response: $other")
  }
}

Java/submit w/invalid response

copysourceCompletableFuture<SubmitResponse> invalidSubmitCommandF =
        hcdCmdService.submit(invalidSetup).thenApply(
                response -> {
                    if (response instanceof Started) {
                        //do something with completed result
                    } else if (response instanceof Invalid invalid) {
                        // Cast the response to get the issue
                        assert (invalid.issue().reason().contains("failure"));
                    }
                    return response;
                }
        );

The handling an immediate completion command looks the same from the command sender’s perspective, but can be challenging on the side of the component handling the command. Because submit returns a Future[SubmitResponse] and onSetup returns SubmitResponse, an immediate completion command must be handled in the component that receives the request. For instance, an Assembly can return a Completed without returning Started, but if an Assembly needs to contact an HCD to get a response, the Assembly must first return Started and then contact the HCD. The following snippet shows a portion of an onSetup handler that returns some local state as an immediate completion.

Scala/submit immediate command

copysourceprivate def onSetup(runId: Id, setup: Setup): SubmitResponse =
  setup.commandName match {
    case `immediateCommand` =>
      val localValue = 1000L
      // Assembly preforms a calculation or reads state information storing in a result
      Completed(runId, Result().add(resultKey.set(localValue)))

Java/submit immediate command

copysourceprivate SubmitResponse onSetup(Id runId, Setup setup) {
  CommandName cmd = setup.commandName();
  if (cmd.equals(immediateCommand)) {
    // Assembly preforms a calculation or reads state information storing in a result
    return new Completed(runId, new Result().add(resultKey.set(1000L)));
  }

submitAndWait

submitAndWait is a convenience method which sends a submit message and then, if the command is long-running, it waits for final completion. Sending a submit message with a command returns a SubmitResponse as a Future. The Future returned by submitAndWait will always be the final response, whether the actions are long-running or not. The final result may be a positive completion (Completed) or a negative completion (Invalid, Error, Cancelled, Locked). The Started response is never seen by the programmer when using submitAndWait of CommandService. The handling of long-running and immediate completion commands look the same from the command sender’s perspective when using submitAndWait.

This example shows an immediate completion command using submitAndWait.

Scala/submitAndWait w/immediate-response

copysourceval immediateSetup = Setup(prefix, immediateCmd, obsId)
val immediateCommandF: Future[SubmitResponse] = async {
  await(assemblyCmdService.submitAndWait(immediateSetup)) match {
    case response: Completed =>
      // do something with completed result
      response
    case otherResponse =>
      // do something with other response which is not expected
      otherResponse
  }
}

Java/submitAndWait w/immediate-response

copysourceCompletableFuture<SubmitResponse> immediateCommandF =
        hcdCmdService.submitAndWait(immediateCmd, timeout).thenApply(
                response -> {
                    if (response instanceof Completed) {
                        //do something with completed result
                    } else {
                        // do something with unexpected response
                    }
                    return response;
                }
        );

Several examples have been shown of sending a command that is long-running. The following examples show an Assembly that issues a command to an HCD with submitAndWait and returns Started. When submitAndWait returns with a final response, the parent is updated with the final response through the CommandResponseManager.

Scala/submitAndWait w/immediate-response

copysourcecase `longRunningCmd` =>
  // A local assembly command that takes some time returning Started
  timeServiceScheduler.scheduleOnce(UTCTime(UTCTime.now().value.plusSeconds(2))) {
    // After time expires, send final update
    commandResponseManager.updateCommand(Completed(runId, Result(encoder.set(20))))
  }
  // Starts long-runing and returns started
  Started(runId)

Java/submitAndWait w/immediate-response

copysourceSetup longRunningSetup = new Setup(prefix(), longRunningCmd(), Optional.empty()).add(encoderValue);
CompletableFuture<Optional<Integer>> longRunningResultF =
        hcdCmdService.submitAndWait(longRunningSetup, timeout)
                .thenCompose(response -> {
                    if (response instanceof Completed) {
                        // This extracts and returns the the first value of parameter encoder
                        Result result = ((Completed) response).result();
                        Optional<Integer> rvalue = Optional.of(result.jGet(encoder).orElseThrow().head());
                        return CompletableFuture.completedFuture(rvalue);
                    } else {
                        // For some other response, return empty
                        return CompletableFuture.completedFuture(Optional.empty());
                    }
                });

oneway

Oneway does not provide completion information but does return the result of validateCommand handler in the Top-Level-Actor (Accepted, Invalid, or Locked). When sending a command as a oneway message, a OnewayResponse is returned as a Future that can be used to check that it was validated if necessary.

Oneway is useful for communication between an Assembly when it needs to send commands to an HCD as quickly as possible. The command is validated on the destination and the validation response is returned, but no other responses are provided.

Scala

copysource// `onewayCmd` is a sample to demonstrate oneway without any actions
val onewaySetup = Setup(prefix, onewayCmd, obsId)
// Don't care about the futures from async
val oneWayF: Future[Unit] = async {
  await(assemblyCmdService.oneway(onewaySetup)) match {
    case invalid: Invalid =>
    // Log an error here
    case _ =>
    // Ignore anything other than invalid
  }
}
oneWayF.map(_ => {
  // do something
})

Java

copysourceSetup onewaySetup = new Setup(prefix(), onewayCmd(), Optional.empty()).add(encoderValue);
CompletableFuture<Void> onewayF = hcdCmdService
        .oneway(onewaySetup)
        .thenAccept(onewayResponse -> {
            if (onewayResponse instanceof Invalid) {
                // log an error here
            } else {
                // Ignore anything other than invalid
            }
        });

validate

Sometimes it may be useful to test whether or not a component can execute a command without committing to executing its actions. The validate message can be used for this purpose. Validate returns a ValidateResponse of Accepted, Invalid, or Locked.

Scala

copysourceval validateCommandF: Future[Boolean] = async {
  await(assemblyCmdService.validate(immediateSetup)) match {
    case _: Accepted       => true
    case Invalid(_, issue) =>
      // do something with other response which is not expected
      log.error(s"Command failed to validate with issue: $issue")
      false
    case _: Locked => false
  }
}

Java

copysourceCompletableFuture<Boolean> validateCommandF =
        hcdCmdService.validate(immediateCmd)
                .thenApply(
                        response -> {
                            if (response instanceof Accepted) {
                                //do something with completed result
                                return true;
                            } else if (response instanceof Invalid) {
                                // do something with unexpected response
                                return false;
                            } else {
                                // Locked
                                return false;
                            }
                        }
                );
Assert.assertTrue(validateCommandF.get());

query

If a command returns a Started response indicating it has long-running actions, and the sender needs to determine that the actions have started properly, or wishes to poll the destination component for the final response, the query method of CommandService can be used as shown in the following example without using the Future returned by submitAndWait, which provides the final completion notification.

Scala/submit w/query

copysourceval longRunningQueryResultF: Future[Option[Int]] = async {
  // The following val is set so we can do query and work and complete later
  val longRunningF = assemblyCmdService.submit(longRunningSetup)
  // This is used in a later test
  longRunningRunId = await(longRunningF).runId

  await(assemblyCmdService.query(longRunningRunId)) match {
    case Started(runId) =>
      runId shouldEqual longRunningRunId
    // happy case - no action needed
    // Do some other work
    case a =>
    // log.error. This indicates that the command probably failed to start.
  }

  // Now wait for completion and result
  await(assemblyCmdService.queryFinal(longRunningRunId)) match {
    case Completed(_, result) =>
      Some(result(encoder).head)

    case otherResponse =>
      // log a message?
      None
  }
}

Java/submit w/query

copysourceCompletableFuture<SubmitResponse> longRunningCommandResultF =
        hcdCmdService.submitAndWait(longRunningSetup, timeout);

// do some work before querying for the result of above command as needed
SubmitResponse sresponse = longRunningCommandResultF.get();
CompletableFuture<SubmitResponse> queryResponseF = hcdCmdService.query(sresponse.runId());
queryResponseF.thenAccept(r -> {
    if (r instanceof Started) {
        // happy case - no action needed
        // Do some other work
    } else {
        // log.error. This indicates that the command probably failed to start.
    }
});

CompletableFuture<Optional<Integer>> intF =
        longRunningCommandResultF.thenCompose(response -> {
            if (response instanceof Completed) {
                // This extracts and returns the the first value of parameter encoder
                Result result = ((Completed) response).result();
                Optional<Integer> rvalue = Optional.of(result.jGet(encoder).orElseThrow().head());
                return CompletableFuture.completedFuture(rvalue);
            } else {
                // For some other response, return empty
                return CompletableFuture.completedFuture(Optional.empty());
            }
        });
Assert.assertEquals(Optional.of(20), intF.get());

queryFinal

When using submit and the submit returns a Started response, the final completion reponse can be obtained using queryFinal. Like query, queryFinal uses the Id returned by Started. But in this case, rather than returning immediately like query, it waits and only returns when the final SubmitResponse is sent. queryFinal is used exclusively with submit in the case where some other activity must be done before the actions started by the submit complete. submitAndWait is just a submit + queryFinal. It is also possible to just start actions with submitAndWait and use the returned Future when you are ready. The following examples show you can use the runId returned by submitAndWait with queryFinal.

Scala/submitAndWait long running/queryFinal

copysourceval encoderValue: Future[Option[Int]] = async {
  // The following submit is made without saving the Future!
  val runId = await(assemblyCmdService.submitAndWait(longRunningSetup)).runId

  // Use queryFinal and runId to wait for completion and result
  await(assemblyCmdService.queryFinal(runId)) match {
    case Completed(_, result) =>
      Some(result(encoder).head)

    case otherResponse =>
      // log a message?
      None
  }
}

Java/submitAndWait long running/queryFinal

copysourcelongRunningCommandResultF = hcdCmdService.submitAndWait(longRunningSetup, timeout);
sresponse = longRunningCommandResultF.get();

// longRunningSetup3 has already been submitted
CompletableFuture<Optional<Integer>> int3F =
        hcdCmdService.queryFinal(sresponse.runId(), timeout).thenCompose(response -> {
            if (response instanceof Completed) {
                // This extracts and returns the the first value of parameter encoder
                Result result = ((Completed) response).result();
                Optional<Integer> rvalue = Optional.of(result.jGet(encoder).orElseThrow().head());
                return CompletableFuture.completedFuture(rvalue);
            } else {
                // For some other response, return empty
                return CompletableFuture.completedFuture(Optional.empty());
            }
        });
Assert.assertEquals(Optional.of(20), int3F.get());

submitAllAndWait

submitAllAndWait can be used to send multiple commands sequentially to the same component. This could be used to send initialization commands to an HCD, for instance. The argument for submitAllAndWait is a list of commands. submitAllAndWait returns a list of SubmitResponses – one for each command in the list. While submitAndWait returns a SubmitResponse as a Future, submitAllAndWait returns a list of SubmitResponses as a future, which completes when all the commands in the list have completed.

Scala/query usage

copysourceval submitAllF: Future[List[SubmitResponse]] = async {
  await(assemblyCmdService.submitAllAndWait(List(submitAllSetup1, submitAllSetup2, submitAllinvalidSetup)))
}
submitAllF.map(submitAllResponse => {
  // do something with submitAllResponse
  submitAllResponse.length // 3
  submitAllResponse.head   // [Completed]
  submitAllResponse(1)     // [Completed]
  submitAllResponse(2)     // [Invalid]
})

Java/query usage

copysourceSetup submitAllSetup1 = new Setup(prefix(), immediateCmd(), Optional.empty()).add(encoderValue);
Setup submitAllSetup2 = new Setup(prefix(), longRunningCmd(), Optional.empty()).add(encoderValue);
Setup submitAllSetup3 = new Setup(prefix(), invalidCmd(), Optional.empty()).add(encoderValue);

CompletableFuture<List<SubmitResponse>> submitAllF = hcdCmdService
        .submitAllAndWait(List.of(submitAllSetup1, submitAllSetup2, submitAllSetup3), timeout);

List<SubmitResponse> submitAllResponse = submitAllF.get();
Assert.assertEquals(submitAllResponse.size(), 3);
Assert.assertTrue(submitAllResponse.get(0) instanceof Completed);
Assert.assertTrue(submitAllResponse.get(1) instanceof Completed);
Assert.assertTrue(submitAllResponse.get(2) instanceof Invalid);

In the first example, three commands are sent and the result is a list with three SubmitResponses. The last one returned invalid and was not executed.

The commands in submitAllAndWait will execute sequentially, but each one must complete successfully for the subsequent commands to be executed. If any one of the commands fails, submitAllAndWait stops and the list is returned with the commands that are completed up to and including the command that failed.

subscribeCurrentState

This method provided by CommandService can be used to subscribe to the CurrentState of a component by providing a callback that is called with the arrival of every CurrentState item. SubscribeCurrentState returns a handle of the CurrentStateSubscription which should be used to unsubscribe the subscription.

CurrentState can be an efficient way to keep the Aseembly up to date with the current internal state of the HCD.

The following example code shows an Assembly that subscribes to all CurrentState items of an HCD. The example sends a Setup with an encoder parameter value to the HCD as a oneway message. In this example, sending this command causes the HCD to publish CurrentState with the value that was sent to it.

Scala

copysource// Subscriber code
val expectedEncoderValue = 234
val currStateSetup       = Setup(prefix, hcdCurrentStateCmd, obsId).add(encoder.set(expectedEncoderValue))
// Setup a callback response to CurrentState
var cstate: CurrentState = CurrentState(prefix, StateName("no cstate"), Set.empty)
val subscription         = hcdCmdService.subscribeCurrentState(cs => cstate = cs)
// Send a oneway to the HCD that will cause it to publish a CurrentState with the encoder value
// in the command parameter "encoder". Callback will store value into cstate.
hcdCmdService.oneway(currStateSetup)

// Eventually current state callback will get invoked when hcd publish its state
// Test to see if value was received
eventually(cstate(encoder).head shouldBe expectedEncoderValue)

// Unsubscribe to CurrentState
subscription.cancel()

Java

copysource// Subscriber cod
int expectedEncoderValue = 234;
Setup currStateSetup = new Setup(prefix(), hcdCurrentStateCmd(), Optional.empty()).add(encoder.set(expectedEncoderValue));
// Setup a callback response to CurrentState - use AtomicInteger to capture final value
final AtomicInteger cstate = new AtomicInteger((1));
Subscription subscription = hcdCmdService.subscribeCurrentState(cs -> {
    // Example sets variable outside scope of closure
    cstate.set(cs.jGet(encoder).orElseThrow().head());
});

// Send a oneway to the HCD that will cause a publish of a CurrentState with the encoder value
// in the command parameter "encoder"
hcdCmdService.oneway(currStateSetup);

// Wait for a bit for the callback
Thread.sleep(200);
// Check to see if CurrentState has the value we sent
Assert.assertEquals(expectedEncoderValue, cstate.get());

// Unsubscribe from CurrentState
subscription.cancel();
// subscribe to the current state of an assembly component and use a callback which forwards each received
// element to a test probe actor
Subscription subscription = hcdCmdService.subscribeCurrentState(currentState -> probe.ref().tell(currentState));

The second part of the example shows the code in the HCD. When the HCD receives the oneway message, it extracts the encoder value and publishes a CurrentState item with the encoder parameter.

Scala

copysource// Publish the CurrentState using parameter set created using a sample Choice parameter
currentStatePublisher.publish(CurrentState(filterAsmPrefix, StateName("testStateName"), Set(choiceKey.set(initChoice))))

Java

copysourceCurrentState initState = currentState.add(SampleComponentState.choiceKey().set(SampleComponentState.initChoice()));
currentStatePublisher.publish(initState);
Key<Integer> encoder = JKeyType.IntKey().make("encoder", JUnits.encoder);
int expectedEncoderValue = setup.jGet(encoder).orElseThrow().head();

CurrentState currentState = new CurrentState(prefix(), new StateName("HCDState")).add(encoder.set(expectedEncoderValue));
currentStatePublisher.publish(currentState);

There are two subscribeCurrentState methods in CommandService. The method shown in the above examples subscribes the caller to all CurrentState published. Each CurrentState item has a StateName. A second signature for subscribeCurrentState can include a Set of StateName when the caller only needs some of the CurrentState published by a component.

Matching State for Command Completion

The matcher is provided to allow a component sending a command to use CurrentState published by a component to determine when actions are complete. The expected case is an Assembly using the CurrentState published by an HCD. When using a submit or submitAndWait, completion is determined in the destination. In some scenarios, the Assembly may want to determine when actions are complete. This is what the matcher allows.

To use this feature, the oneway message is used rather than submit. A oneway command is validated but the framework does not provide completion. Doing a query with the runId of a oneway will always return CommandNotAvailable, but oneway is perfect for use with a matcher.

The matcher is created with the ActorRef of the component that is the source of CurrentState and an instance of StateMatcher, which defines the state and criteria for matching.

Several types of StateMatcher are provided as part of CSW for common use. These are DemandMatcherAll for matching the entire DemandState against the current state, DemandMatcher for matching state with or without units against the current state, and PresenceMatcher which checks if a matching state is found with a provided prefix.

The developer is not limited to these StateMatchers. Any class the implements the StateMatcher interface can be provided to a Matcher.

Scala

copysourceval param: Parameter[Int] = encoder.set(100)
val setupWithMatcher      = Setup(prefix, matcherCmd, obsId)

// create a StateMatcher which specifies the desired algorithm and state to be matched.
val demandMatcher: StateMatcher =
  DemandMatcher(DemandState(prefix, StateName("testStateName")).add(param), withUnits = false, timeout)

// Submit command as a oneway and if the command is successfully validated,
// check for matching of demand state against current state
val matchResponseF: Future[MatchingResponse] = assemblyCmdService.onewayAndMatch(setupWithMatcher, demandMatcher)
matchResponseF.map(commandResponse => {
  commandResponse // [Completed]
})

Java

copysourceParameter<Integer> param = JKeyType.IntKey().make("encoder", JUnits.encoder).set(100);
Setup setupWithMatcher = new Setup(prefix(), matcherCmd(), Optional.empty()).add(param);

// create a StateMatcher which specifies the desired algorithm and state to be matched.
DemandMatcher demandMatcher = new DemandMatcher(new DemandState(prefix(), new StateName("testStateName")).add(param), false, timeout);

// Submit command as a oneway and if the command is successfully validated,
// check for matching of demand state against current state
CompletableFuture<MatchingResponse> matchResponseF = hcdCmdService.onewayAndMatch(setupWithMatcher, demandMatcher);
MatchingResponse actualResponse = matchResponseF.get();
Completed expectedResponse = new Completed(actualResponse.runId());
Assert.assertEquals(expectedResponse, actualResponse);            // Not a great test for now

One important point is that the matcher is created and must be shutdown when you are finished with it using the stop method of the matcher as shown in the example.

onewayAndMatch

CommandService provides a short-cut called onewayAndMatch that combines a oneway and a matcher and implements much of the boilerplate of the previous example.

Scala

copysourceval onewayMatchF: Future[SubmitResponse & MatchingResponse] = async {
  await(assemblyCmdService.onewayAndMatch(setupWithMatcher, demandMatcher)) match {
    case i: Invalid =>
      // Command was not accepted
      log.error(s"Oneway match was not accepted: ${i.issue}")
      i
    case c: Completed =>
      // Do some completed work
      c
    case e: Error =>
      // Match failed and timedout generating an error - log a message
      log.error(s"Oeway match produced an error: ${e.message}")
      e
    case l: Locked =>
      // Destination component was locked, log a message
      log.error(s"Destination component was locked")
      l
  }
}

Java

copysource// create a DemandMatcher which specifies the desired state to be matched.
StateMatcher stateMatcher = new DemandMatcher(new DemandState(prefix(), new StateName("testStateName")).add(param), false, timeout);

CompletableFuture<MatchingResponse> matchedCommandResponseF =
        hcdCmdService.onewayAndMatch(setupWithMatcher, stateMatcher);