PusherStreamingService extends ConnectableService { public Observable<String> subscribeChannel(String channelName, String eventName) { return subscribeChannel(channelName, Collections.singletonList(eventName)); } PusherStreamingService(String apiKey);  PusherStreamingService(String apiKey, String cluster);  PusherStreamingService(String apiKey, PusherOptions options); protected  PusherStreamingService(Pusher pusher); Completable disconnect(); Observable<String> subscribePrivateChannel(String channelName, String eventName); Observable<String> subscribeChannel(String channelName, String eventName); Observable<String> subscribeChannel(String channelName, List<String> eventsName); Observable<String> subscribePrivateChannel(String channelName, List<String> eventsName); boolean isSocketOpen(); void useCompressedMessages(boolean compressedMessages);  }
@Test public void testSubscribeChannel() throws Exception { when(pusher.getConnection()).thenReturn(connection); when(connection.getState()).thenReturn(ConnectionState.CONNECTED); when(pusher.subscribe(any())).thenReturn(channel); TestObserver<String> test = streamingService.subscribeChannel("channelName", "eventName").test(); test.assertNoErrors(); test.assertNotTerminated(); verify(pusher).subscribe("channelName"); ArgumentCaptor<SubscriptionEventListener> subscription = ArgumentCaptor.forClass(SubscriptionEventListener.class); verify(channel).bind(anyString(), subscription.capture()); Map<String, Object> pusherEventMap = new HashMap<>(); pusherEventMap.put("data", "dataObject"); subscription.getValue().onEvent(new PusherEvent(pusherEventMap)); test.assertValue("dataObject"); }
@Test public void testSubscribeChannelNotConnected() throws Exception { when(pusher.getConnection()).thenReturn(connection); when(connection.getState()).thenReturn(ConnectionState.DISCONNECTED); TestObserver<String> test = streamingService.subscribeChannel("channelName", "eventName").test(); test.assertError(NotConnectedException.class); verify(pusher, never()).subscribe("channelName"); }