### Question:
MqttProtocolClient implements ProtocolClient { @Override public Observable<Content> observeResource(Form form) throws ProtocolClientException { try { String topic = new URI(form.getHref()).getPath().substring(1); return topicSubjects.computeIfAbsent(topic, key -> topicObserver(form, topic)); } catch (URISyntaxException e) { throw new ProtocolClientException("Unable to subscribe resource: " + e.getMessage()); } } MqttProtocolClient(Pair<MqttProtocolSettings, MqttClient> settingsClientPair);   MqttProtocolClient(Pair<MqttProtocolSettings, MqttClient> settingsClientPair,
                       Map<String, Observable<Content>> topicSubjects); @Override CompletableFuture<Content> invokeResource(Form form); @Override CompletableFuture<Content> invokeResource(Form form, Content content); @Override Observable<Content> observeResource(Form form); @Override Observable<Thing> discover(ThingFilter filter);  }### Answer:
@Test public void subscribeResourceShouldSubscribeToBroker() throws MqttException, ProtocolClientException { when(form.getHref()).thenReturn("tcp: when(content.getBody()).thenReturn("Hallo Welt".getBytes()); client = new MqttProtocolClient(new Pair(settings, mqttClient), new HashMap<>()); client.observeResource(form).subscribe(); verify(mqttClient).subscribe(eq("counter/events/change"), any()); }
@Test public void subscribeResourceShouldInformObserverAboutNextValue() throws ProtocolClientException, MqttException, ContentCodecException, ExecutionException, InterruptedException, TimeoutException { when(form.getHref()).thenReturn("tcp: when(form.getContentType()).thenReturn("application/json"); when(content.getBody()).thenReturn("Hallo Welt".getBytes()); doAnswer(new AnswersWithDelay(1 * 1000L, invocation -> { String topic = invocation.getArgument(0, String.class); IMqttMessageListener listener = invocation.getArgument(1, IMqttMessageListener.class); listener.messageArrived(topic, new MqttMessage("\"Hallo Welt\"".getBytes())); return null; })).when(mqttClient).subscribe(any(String.class), any(IMqttMessageListener.class)); client = new MqttProtocolClient(new Pair(settings, mqttClient), new HashMap<>()); await().untilAsserted(() -> assertEquals( ContentManager.valueToContent("Hallo Welt"), client.observeResource(form).firstElement().blockingGet() )); }
@Test public void subscribeResourceShouldReuseExistingBrokerSubscriptions() throws ProtocolClientException { when(form.getHref()).thenReturn("tcp: when(content.getBody()).thenReturn("Hallo Welt".getBytes()); LambdaObserver<Content> observer = new LambdaObserver<>(n -> { }, e -> { }, () -> { }, d -> { }); client = new MqttProtocolClient(new Pair(settings, mqttClient), new HashMap(Map.of("counter/events/change", subject))); client.observeResource(form).subscribe(observer); verify(subject).subscribe(observer); }
@Test public void subscribeResourceShouldUnsubscribeFromBrokerWhenSubscriptionIsNotLongUsed() throws MqttException, ProtocolClientException { when(form.getHref()).thenReturn("tcp: when(content.getBody()).thenReturn("Hallo Welt".getBytes()); LambdaObserver<Content> observer = new LambdaObserver<>(n -> { }, e -> { }, () -> { }, d -> { }); client = new MqttProtocolClient(new Pair(settings, mqttClient), new HashMap()); client.observeResource(form).subscribe(observer); observer.dispose(); verify(mqttClient).unsubscribe(eq("counter/events/change")); }