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.
Note that the Scala examples are using async/await which simplifies handling the Futures, but is not necessary. The async/await library is not available in Java.
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
-
source
class 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
-
source
private 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
-
source
private 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
-
source
private 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
-
source
val 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
-
source
CompletableFuture<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
-
source
private 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
-
source
private 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
-
source
val 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
-
source
CompletableFuture<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
-
source
case `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
-
source
Setup 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
-
source
// `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
-
source
Setup 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
-
source
val 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
-
source
CompletableFuture<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
-
source
val 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
-
source
CompletableFuture<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
-
source
val 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
-
source
longRunningCommandResultF = 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 SubmitResponse
s – one for each command in the list. While submitAndWait
returns a SubmitResponse
as a Future, submitAllAndWait
returns a list of SubmitResponse
s as a future, which completes when all the commands in the list have completed.
- Scala/query usage
-
source
val 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
-
source
Setup 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 SubmitResponse
s. 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.
Callbacks are not thread-safe on the JVM. If you are doing side effects/mutations inside the callback, you should ensure that it is done in a thread-safe way inside an actor. Here is an example of how it can be done.
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
-
source
// 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
-
source
// 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
-
source
// Publish the CurrentState using parameter set created using a sample Choice parameter currentStatePublisher.publish(CurrentState(filterAsmPrefix, StateName("testStateName"), Set(choiceKey.set(initChoice))))
- Java
-
source
CurrentState 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 StateMatcher
s. Any class the implements the StateMatcher
interface can be provided to a Matcher
.
- Scala
-
source
val 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
-
source
Parameter<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
-
source
val 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
-
source
// 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);