@Test public void blowsUpWhenEventIsSagaStarted() { TxEvent event = new SagaStartedEvent(globalTxId, localTxId, 0); try { messageSender.send(event); expectFailing(OmegaException.class); } catch (OmegaException e) { assertThat(e.getMessage(), is("Failed to process subsequent requests because no alpha server is available")); } }
@Override public AlphaResponse send(TxEvent event) { if (event.type() == SagaStartedEvent) { throw new OmegaException("Failed to process subsequent requests because no alpha server is available"); } try { return availableMessageSenders.take().send(event); } catch (InterruptedException e) { throw new OmegaException("Failed to send event " + event + " due to interruption", e); } catch (Exception e) { throw e; } }
RetryableMessageSender implements MessageSender { @Override public AlphaResponse send(TxEvent event) { if (event.type() == SagaStartedEvent) { throw new OmegaException("Failed to process subsequent requests because no alpha server is available"); } try { return availableMessageSenders.take().send(event); } catch (InterruptedException e) { throw new OmegaException("Failed to send event " + event + " due to interruption", e); } catch (Exception e) { throw e; } } }
RetryableMessageSender implements MessageSender { @Override public AlphaResponse send(TxEvent event) { if (event.type() == SagaStartedEvent) { throw new OmegaException("Failed to process subsequent requests because no alpha server is available"); } try { return availableMessageSenders.take().send(event); } catch (InterruptedException e) { throw new OmegaException("Failed to send event " + event + " due to interruption", e); } catch (Exception e) { throw e; } } RetryableMessageSender(BlockingQueue<MessageSender> availableMessageSenders); }
RetryableMessageSender implements MessageSender { @Override public AlphaResponse send(TxEvent event) { if (event.type() == SagaStartedEvent) { throw new OmegaException("Failed to process subsequent requests because no alpha server is available"); } try { return availableMessageSenders.take().send(event); } catch (InterruptedException e) { throw new OmegaException("Failed to send event " + event + " due to interruption", e); } catch (Exception e) { throw e; } } RetryableMessageSender(BlockingQueue<MessageSender> availableMessageSenders); @Override void onConnected(); @Override void onDisconnected(); @Override void close(); @Override String target(); @Override AlphaResponse send(TxEvent event); @Override Set<String> send(Set<String> localTxIdSet); @Override String reportMessageToServer(KafkaMessage message); @Override String reportAccidentToServer(AccidentHandling accidentHandling); @Override GrpcConfigAck readConfigFromServer(int type, String category); }
RetryableMessageSender implements MessageSender { @Override public AlphaResponse send(TxEvent event) { if (event.type() == SagaStartedEvent) { throw new OmegaException("Failed to process subsequent requests because no alpha server is available"); } try { return availableMessageSenders.take().send(event); } catch (InterruptedException e) { throw new OmegaException("Failed to send event " + event + " due to interruption", e); } catch (Exception e) { throw e; } } RetryableMessageSender(BlockingQueue<MessageSender> availableMessageSenders); @Override void onConnected(); @Override void onDisconnected(); @Override void close(); @Override String target(); @Override AlphaResponse send(TxEvent event); @Override Set<String> send(Set<String> localTxIdSet); @Override String reportMessageToServer(KafkaMessage message); @Override String reportAccidentToServer(AccidentHandling accidentHandling); @Override GrpcConfigAck readConfigFromServer(int type, String category);  }