@Test public void pushFailedCallbackToEndOfQueue() throws Exception { doThrow(RuntimeException.class).doThrow(RuntimeException.class).doNothing().when(sender).onDisconnected(); assertThat(runnables, contains(NO_OP_RUNNABLE)); pushBack.run(); assertThat(runnables, contains(NO_OP_RUNNABLE, pushBack)); assertThat(runnables.poll(), is(NO_OP_RUNNABLE)); assertThat(runnables.contains(pushBack), is(true)); runnables.poll().run(); assertThat(runnables.contains(pushBack), is(true)); runnables.poll().run(); assertThat(runnables.isEmpty(), is(true)); assertThat(senders.get(sender), is(0L)); assertThat(connectedSenders, contains(sender)); verify(sender, times(3)).onDisconnected(); verify(sender, times(1)).onConnected(); }
@Override public void run() { try { LOG.info("Retry connecting to alpha at {}", messageSender.target()); messageSender.onDisconnected(); messageSender.onConnected(); senders.put(messageSender, 0L); connectedSenders.offer(messageSender); LOG.info("Retry connecting to alpha at {} is successful", messageSender.target()); } catch (Exception e) { LOG.error("Failed to reconnect to alpha at {}", messageSender.target(), e); pendingTasks.offer(this); } }
PushBackReconnectRunnable implements Runnable { @Override public void run() { try { LOG.info("Retry connecting to alpha at {}", messageSender.target()); messageSender.onDisconnected(); messageSender.onConnected(); senders.put(messageSender, 0L); connectedSenders.offer(messageSender); LOG.info("Retry connecting to alpha at {} is successful", messageSender.target()); } catch (Exception e) { LOG.error("Failed to reconnect to alpha at {}", messageSender.target(), e); pendingTasks.offer(this); } } }
PushBackReconnectRunnable implements Runnable { @Override public void run() { try { LOG.info("Retry connecting to alpha at {}", messageSender.target()); messageSender.onDisconnected(); messageSender.onConnected(); senders.put(messageSender, 0L); connectedSenders.offer(messageSender); LOG.info("Retry connecting to alpha at {} is successful", messageSender.target()); } catch (Exception e) { LOG.error("Failed to reconnect to alpha at {}", messageSender.target(), e); pendingTasks.offer(this); } } PushBackReconnectRunnable(
      MessageSender messageSender,
      Map<MessageSender, Long> senders,
      BlockingQueue<Runnable> pendingTasks,
      BlockingQueue<MessageSender> connectedSenders); }
PushBackReconnectRunnable implements Runnable { @Override public void run() { try { LOG.info("Retry connecting to alpha at {}", messageSender.target()); messageSender.onDisconnected(); messageSender.onConnected(); senders.put(messageSender, 0L); connectedSenders.offer(messageSender); LOG.info("Retry connecting to alpha at {} is successful", messageSender.target()); } catch (Exception e) { LOG.error("Failed to reconnect to alpha at {}", messageSender.target(), e); pendingTasks.offer(this); } } PushBackReconnectRunnable(
      MessageSender messageSender,
      Map<MessageSender, Long> senders,
      BlockingQueue<Runnable> pendingTasks,
      BlockingQueue<MessageSender> connectedSenders); @Override void run(); }
PushBackReconnectRunnable implements Runnable { @Override public void run() { try { LOG.info("Retry connecting to alpha at {}", messageSender.target()); messageSender.onDisconnected(); messageSender.onConnected(); senders.put(messageSender, 0L); connectedSenders.offer(messageSender); LOG.info("Retry connecting to alpha at {} is successful", messageSender.target()); } catch (Exception e) { LOG.error("Failed to reconnect to alpha at {}", messageSender.target(), e); pendingTasks.offer(this); } } PushBackReconnectRunnable(
      MessageSender messageSender,
      Map<MessageSender, Long> senders,
      BlockingQueue<Runnable> pendingTasks,
      BlockingQueue<MessageSender> connectedSenders); @Override void run();  }