### Question:
AkkaProtocolClient implements ProtocolClient { @Override public Observable<Content> observeResource(Form form) { return observe(form, resourceIdentifier -> { if (resourceIdentifier.length == 2 && resourceIdentifier[0].equals("properties")) { return new SubscribeProperty(resourceIdentifier[1]); } else if (resourceIdentifier.length == 2 && resourceIdentifier[0].equals("events")) { return new SubscribeEvent(resourceIdentifier[1]); } else { return null; } }); } AkkaProtocolClient(ActorSystem system,
                              Duration askTimeout,
                              Duration discoverTimeout);   AkkaProtocolClient(ActorSystem system,
                       Duration askTimeout,
                       Duration discoverTimeout,
                       AkkaProtocolPattern pattern); @SuppressWarnings("squid:S1192") @Override CompletableFuture<Content> readResource(Form form); @Override CompletableFuture<Content> writeResource(Form form, Content content); @Override CompletableFuture<Content> invokeResource(Form form, Content content); @Override Observable<Content> observeResource(Form form); @Override Observable<Thing> discover(ThingFilter filter);  }### Answer:
@Test public void observeResourceShouldCreateCorrectActorForProperty() { when(form.getHref()).thenReturn("akka: when(system.actorSelection(anyString())).thenReturn(actorSelection); when(pattern.ask(any(ActorSelection.class), any(), any())).thenReturn(completedFuture(message)); AkkaProtocolClient client = new AkkaProtocolClient(system, askTimeout, discoverTimeout, pattern); client.observeResource(form).subscribe(); verify(system).actorOf(any()); }
@Test public void observeResourceShouldCreateCorrectActorForEvent() { when(form.getHref()).thenReturn("akka: when(system.actorSelection(anyString())).thenReturn(actorSelection); when(pattern.ask(any(ActorSelection.class), any(), any())).thenReturn(completedFuture(message)); AkkaProtocolClient client = new AkkaProtocolClient(system, askTimeout, discoverTimeout, pattern); client.observeResource(form).subscribe(); verify(system).actorOf(any()); }
@Test public void subscribeResourceShouldStopActorWhenObserverIsDone() { when(form.getHref()).thenReturn("akka: AkkaProtocolClient client = new AkkaProtocolClient(system, askTimeout, discoverTimeout, pattern); client.observeResource(form).subscribe().dispose(); verify(system).stop(any()); }