@Test(groups = "fast") public void testSend() throws Exception { int i = 100; while (i > 0) { i--; } }
@Override public void send(final File file, final CallbackHandler handler) { if (isClosed.get()) { createConnection(); } sleeping.set(false); final List<LogEntry> list = createScribePayload(file, handler); if (list == null) { return; } try { scribeClient.log(list); handler.onSuccess(file); messagesSuccessfullySent.addAndGet(list.size()); messagesSuccessfullySentSinceLastReconnection.addAndGet(list.size()); if (messagesSuccessfullySentSinceLastReconnection.get() > messagesToSendBeforeReconnecting) { log.info("Recycling connection with Scribe"); messagesSuccessfullySentSinceLastReconnection.set(0); createConnection(); } } catch (org.apache.thrift.TException e) { log.warn("Error while sending message to Scribe: {}", e.getLocalizedMessage()); createConnection(); handler.onError(new Throwable(e), file); } }
ScribeSender implements EventSender { @Override public void send(final File file, final CallbackHandler handler) { if (isClosed.get()) { createConnection(); } sleeping.set(false); final List<LogEntry> list = createScribePayload(file, handler); if (list == null) { return; } try { scribeClient.log(list); handler.onSuccess(file); messagesSuccessfullySent.addAndGet(list.size()); messagesSuccessfullySentSinceLastReconnection.addAndGet(list.size()); if (messagesSuccessfullySentSinceLastReconnection.get() > messagesToSendBeforeReconnecting) { log.info("Recycling connection with Scribe"); messagesSuccessfullySentSinceLastReconnection.set(0); createConnection(); } } catch (org.apache.thrift.TException e) { log.warn("Error while sending message to Scribe: {}", e.getLocalizedMessage()); createConnection(); handler.onError(new Throwable(e), file); } } }
ScribeSender implements EventSender { @Override public void send(final File file, final CallbackHandler handler) { if (isClosed.get()) { createConnection(); } sleeping.set(false); final List<LogEntry> list = createScribePayload(file, handler); if (list == null) { return; } try { scribeClient.log(list); handler.onSuccess(file); messagesSuccessfullySent.addAndGet(list.size()); messagesSuccessfullySentSinceLastReconnection.addAndGet(list.size()); if (messagesSuccessfullySentSinceLastReconnection.get() > messagesToSendBeforeReconnecting) { log.info("Recycling connection with Scribe"); messagesSuccessfullySentSinceLastReconnection.set(0); createConnection(); } } catch (org.apache.thrift.TException e) { log.warn("Error while sending message to Scribe: {}", e.getLocalizedMessage()); createConnection(); handler.onError(new Throwable(e), file); } } ScribeSender(final ScribeClient scribeClient, final int messagesToSendBeforeReconnecting, final int maxIdleTimeInMinutes); }
ScribeSender implements EventSender { @Override public void send(final File file, final CallbackHandler handler) { if (isClosed.get()) { createConnection(); } sleeping.set(false); final List<LogEntry> list = createScribePayload(file, handler); if (list == null) { return; } try { scribeClient.log(list); handler.onSuccess(file); messagesSuccessfullySent.addAndGet(list.size()); messagesSuccessfullySentSinceLastReconnection.addAndGet(list.size()); if (messagesSuccessfullySentSinceLastReconnection.get() > messagesToSendBeforeReconnecting) { log.info("Recycling connection with Scribe"); messagesSuccessfullySentSinceLastReconnection.set(0); createConnection(); } } catch (org.apache.thrift.TException e) { log.warn("Error while sending message to Scribe: {}", e.getLocalizedMessage()); createConnection(); handler.onError(new Throwable(e), file); } } ScribeSender(final ScribeClient scribeClient, final int messagesToSendBeforeReconnecting, final int maxIdleTimeInMinutes); synchronized void createConnection(); @Override synchronized void close(); @Override void send(final File file, final CallbackHandler handler); @Managed(description = "Get the number of messages successfully sent since startup to Scribe") long getMessagesSuccessfullySent(); @Managed(description = "Get the number of messages successfully sent since last reconnection to Scribe") long getMessagesSuccessfullySentSinceLastReconnection(); @Managed(description = "Get the number of times we retried to connect to Scribe") long getConnectionRetries(); }
ScribeSender implements EventSender { @Override public void send(final File file, final CallbackHandler handler) { if (isClosed.get()) { createConnection(); } sleeping.set(false); final List<LogEntry> list = createScribePayload(file, handler); if (list == null) { return; } try { scribeClient.log(list); handler.onSuccess(file); messagesSuccessfullySent.addAndGet(list.size()); messagesSuccessfullySentSinceLastReconnection.addAndGet(list.size()); if (messagesSuccessfullySentSinceLastReconnection.get() > messagesToSendBeforeReconnecting) { log.info("Recycling connection with Scribe"); messagesSuccessfullySentSinceLastReconnection.set(0); createConnection(); } } catch (org.apache.thrift.TException e) { log.warn("Error while sending message to Scribe: {}", e.getLocalizedMessage()); createConnection(); handler.onError(new Throwable(e), file); } } ScribeSender(final ScribeClient scribeClient, final int messagesToSendBeforeReconnecting, final int maxIdleTimeInMinutes); synchronized void createConnection(); @Override synchronized void close(); @Override void send(final File file, final CallbackHandler handler); @Managed(description = "Get the number of messages successfully sent since startup to Scribe") long getMessagesSuccessfullySent(); @Managed(description = "Get the number of messages successfully sent since last reconnection to Scribe") long getMessagesSuccessfullySentSinceLastReconnection(); @Managed(description = "Get the number of times we retried to connect to Scribe") long getConnectionRetries();  }