@Test public void pushFailedCallbackToEndOfQueue() throws Exception { TxEvent event = someEvent(); doThrow(AlphaException.class).doThrow(AlphaException.class).doNothing().when(underlying).compensate(event); pushBack.compensate(event); assertThat(runnables.size(), is(2)); assertThat(runnables.poll(), is(NO_OP_RUNNABLE)); runnables.poll().run(); assertThat(runnables.size(), is(1)); runnables.poll().run(); verify(underlying, times(3)).compensate(event); }
@Override public void compensate(TxEvent event) { try { underlying.compensate(event); } catch (Exception e) { logError(event, e); pendingCompensations.offer(() -> compensate(event)); } }
PushBackOmegaCallback implements OmegaCallback { @Override public void compensate(TxEvent event) { try { underlying.compensate(event); } catch (Exception e) { logError(event, e); pendingCompensations.offer(() -> compensate(event)); } } }
PushBackOmegaCallback implements OmegaCallback { @Override public void compensate(TxEvent event) { try { underlying.compensate(event); } catch (Exception e) { logError(event, e); pendingCompensations.offer(() -> compensate(event)); } } PushBackOmegaCallback(BlockingQueue<Runnable> pendingCompensations, OmegaCallback underlying); }
PushBackOmegaCallback implements OmegaCallback { @Override public void compensate(TxEvent event) { try { underlying.compensate(event); } catch (Exception e) { logError(event, e); pendingCompensations.offer(() -> compensate(event)); } } PushBackOmegaCallback(BlockingQueue<Runnable> pendingCompensations, OmegaCallback underlying); @Override void compensate(TxEvent event); }
PushBackOmegaCallback implements OmegaCallback { @Override public void compensate(TxEvent event) { try { underlying.compensate(event); } catch (Exception e) { logError(event, e); pendingCompensations.offer(() -> compensate(event)); } } PushBackOmegaCallback(BlockingQueue<Runnable> pendingCompensations, OmegaCallback underlying); @Override void compensate(TxEvent event);  }