### Question:
Socket { public Observable<Object> connection() { return connection; } Socket(@Nonnull SocketConnection socketConnection, @Nonnull Scheduler scheduler); Observable<RxObjectEvent> events(); Observable<RxObjectEventConn> connectedAndRegistered(); Observable<Object> connection(); void sendPingWhenConnected(); void sendPingEvery5seconds(); @Nonnull Observable<String> nextId(); @Nonnull Observable<DataMessage> sendMessageOnceWhenConnected(final Func1<String, Observable<Object>> createMessage); static final Logger LOGGER; }### Answer:
@Test public void testConnection_registerIsSent() throws Exception { final Subscription subscribe = socket.connection().subscribe(observer); try { connection.onNext(new RxObjectEventConnected(sender), 0); testScheduler.triggerActions(); verify(sender).sendObjectMessage(new RegisterMessage("asdf")); } finally { subscribe.unsubscribe(); } }
@Test public void testAfterConnection_registerSuccess() throws Exception { final Subscription subscribe = socket.connection().subscribe(observer); try { register(); } finally { subscribe.unsubscribe(); } }
@Test public void testConnectionSuccessAfterAWhile_registerSuccess() throws Exception { final Subscription subscribe = socket.connection().subscribe(observer); try { testScheduler.advanceTimeBy(30, TimeUnit.SECONDS); register(); } finally { subscribe.unsubscribe(); } }