@Test public void willDisconnectOnEndStream() { setupWebserverConnection(); sendFirstMessage(); webSocketChannelServer.sendEndStream(); verify(webSocketConnection).disconnect(); }
@Override public boolean sendEndStream() { disconnectedWithEndStreamCall = true; sendMessageQueue.onNext(CLOSE_MESSAGE); Observable.timer(WAIT_FOR_CLOSE_TIMEOUT, TimeUnit.SECONDS, getSendScheduler()) .subscribe(new Consumer<Long>() { @Override public void accept(Long aLong) throws Exception { if (!sendMessageQueue.hasComplete()) { sendMessageQueue.onComplete(); } } }); return true; }
WebSocketChannelServer extends MessengerChannelServer { @Override public boolean sendEndStream() { disconnectedWithEndStreamCall = true; sendMessageQueue.onNext(CLOSE_MESSAGE); Observable.timer(WAIT_FOR_CLOSE_TIMEOUT, TimeUnit.SECONDS, getSendScheduler()) .subscribe(new Consumer<Long>() { @Override public void accept(Long aLong) throws Exception { if (!sendMessageQueue.hasComplete()) { sendMessageQueue.onComplete(); } } }); return true; } }
WebSocketChannelServer extends MessengerChannelServer { @Override public boolean sendEndStream() { disconnectedWithEndStreamCall = true; sendMessageQueue.onNext(CLOSE_MESSAGE); Observable.timer(WAIT_FOR_CLOSE_TIMEOUT, TimeUnit.SECONDS, getSendScheduler()) .subscribe(new Consumer<Long>() { @Override public void accept(Long aLong) throws Exception { if (!sendMessageQueue.hasComplete()) { sendMessageQueue.onComplete(); } } }); return true; } WebSocketChannelServer(Context context, String serviceComponentName, String clientPackageName); }
WebSocketChannelServer extends MessengerChannelServer { @Override public boolean sendEndStream() { disconnectedWithEndStreamCall = true; sendMessageQueue.onNext(CLOSE_MESSAGE); Observable.timer(WAIT_FOR_CLOSE_TIMEOUT, TimeUnit.SECONDS, getSendScheduler()) .subscribe(new Consumer<Long>() { @Override public void accept(Long aLong) throws Exception { if (!sendMessageQueue.hasComplete()) { sendMessageQueue.onComplete(); } } }); return true; } WebSocketChannelServer(Context context, String serviceComponentName, String clientPackageName); @Override void handleMessage(Message msg); @Override boolean send(final String message); @Override boolean sendEndStream(); }
WebSocketChannelServer extends MessengerChannelServer { @Override public boolean sendEndStream() { disconnectedWithEndStreamCall = true; sendMessageQueue.onNext(CLOSE_MESSAGE); Observable.timer(WAIT_FOR_CLOSE_TIMEOUT, TimeUnit.SECONDS, getSendScheduler()) .subscribe(new Consumer<Long>() { @Override public void accept(Long aLong) throws Exception { if (!sendMessageQueue.hasComplete()) { sendMessageQueue.onComplete(); } } }); return true; } WebSocketChannelServer(Context context, String serviceComponentName, String clientPackageName); @Override void handleMessage(Message msg); @Override boolean send(final String message); @Override boolean sendEndStream(); static final String CONNECT_PLEASE; static final String CLOSE_MESSAGE; }