@Test public void testSendReturnsFalseOnScheduleSend() throws Exception { class StubTelemetriesTransmitter implements TelemetriesTransmitter<String> { private int scheduleSendCounter = 2; private Collection<String> sendNowCollection; @Override public boolean scheduleSend(TelemetriesFetcher telemetriesFetcher, long value, TimeUnit timeUnit) { --scheduleSendCounter; if (scheduleSendCounter > 0) { return false; } return true; } @Override public boolean sendNow(Collection<String> telemetries) { sendNowCollection = telemetries; return true; } @Override public void shutdown(long timeout, TimeUnit timeUnit) { } public Collection<String> getSendNowCollection() { return sendNowCollection; } }; List<String> all = new ArrayList<String>(); List<String> expected = new ArrayList<String>(); for (int i = 0; i < 4; ++i) { String mockSerializedTelemetry = "mockTelemtry" + String.valueOf(i); all.add(mockSerializedTelemetry); if (i != 0) { expected.add(mockSerializedTelemetry); } } StubTelemetriesTransmitter mockSender = new StubTelemetriesTransmitter(); LimitsEnforcer maxEnforcer = createEnforcerWithCurrentValue(3); LimitsEnforcer sendEnforcer = createDefaultSenderTimeoutEnforcer(); TelemetryBuffer testedBuffer = new TelemetryBuffer(mockSender, maxEnforcer, sendEnforcer); for (String telemetry : all) { testedBuffer.add(telemetry); } Collection<String> sendNowCollection = mockSender.getSendNowCollection(); assertEquals(sendNowCollection.size(), expected.size()); int i = 0; for (String telemetry : sendNowCollection) { assertEquals(telemetry, expected.get(i)); ++i; } }
public void add(T telemetry) { Preconditions.checkNotNull(telemetry, "Telemetry must be non null value"); synchronized (lock) { telemetries.add(telemetry); int currentSize = telemetries.size(); if (currentSize >= maxTelemetriesInBatch) { if (!sender.sendNow(prepareTelemetriesForSend())) { logger.error("Failed to send buffer data to network"); } } else if (currentSize == 1) { if (!sender.scheduleSend(new TelemetryBufferTelemetriesFetcher(generation), transmitBufferTimeoutInSeconds, TimeUnit.SECONDS)) { logger.error("Failed to schedule send of the buffer to network"); telemetries.clear(); } } } }
TelemetryBuffer { public void add(T telemetry) { Preconditions.checkNotNull(telemetry, "Telemetry must be non null value"); synchronized (lock) { telemetries.add(telemetry); int currentSize = telemetries.size(); if (currentSize >= maxTelemetriesInBatch) { if (!sender.sendNow(prepareTelemetriesForSend())) { logger.error("Failed to send buffer data to network"); } } else if (currentSize == 1) { if (!sender.scheduleSend(new TelemetryBufferTelemetriesFetcher(generation), transmitBufferTimeoutInSeconds, TimeUnit.SECONDS)) { logger.error("Failed to schedule send of the buffer to network"); telemetries.clear(); } } } } }
TelemetryBuffer { public void add(T telemetry) { Preconditions.checkNotNull(telemetry, "Telemetry must be non null value"); synchronized (lock) { telemetries.add(telemetry); int currentSize = telemetries.size(); if (currentSize >= maxTelemetriesInBatch) { if (!sender.sendNow(prepareTelemetriesForSend())) { logger.error("Failed to send buffer data to network"); } } else if (currentSize == 1) { if (!sender.scheduleSend(new TelemetryBufferTelemetriesFetcher(generation), transmitBufferTimeoutInSeconds, TimeUnit.SECONDS)) { logger.error("Failed to schedule send of the buffer to network"); telemetries.clear(); } } } } TelemetryBuffer(TelemetriesTransmitter<T> sender, LimitsEnforcer maxTelemetriesInBatchEnforcer, LimitsEnforcer transmitBufferTimeoutInSecondsEnforcer); }
TelemetryBuffer { public void add(T telemetry) { Preconditions.checkNotNull(telemetry, "Telemetry must be non null value"); synchronized (lock) { telemetries.add(telemetry); int currentSize = telemetries.size(); if (currentSize >= maxTelemetriesInBatch) { if (!sender.sendNow(prepareTelemetriesForSend())) { logger.error("Failed to send buffer data to network"); } } else if (currentSize == 1) { if (!sender.scheduleSend(new TelemetryBufferTelemetriesFetcher(generation), transmitBufferTimeoutInSeconds, TimeUnit.SECONDS)) { logger.error("Failed to schedule send of the buffer to network"); telemetries.clear(); } } } } TelemetryBuffer(TelemetriesTransmitter<T> sender, LimitsEnforcer maxTelemetriesInBatchEnforcer, LimitsEnforcer transmitBufferTimeoutInSecondsEnforcer); void setMaxTelemetriesInBatch(int value); int getMaxTelemetriesInBatch(); void setTransmitBufferTimeoutInSeconds(int value); int getTransmitBufferTimeoutInSeconds(); void add(T telemetry); void flush(); }
TelemetryBuffer { public void add(T telemetry) { Preconditions.checkNotNull(telemetry, "Telemetry must be non null value"); synchronized (lock) { telemetries.add(telemetry); int currentSize = telemetries.size(); if (currentSize >= maxTelemetriesInBatch) { if (!sender.sendNow(prepareTelemetriesForSend())) { logger.error("Failed to send buffer data to network"); } } else if (currentSize == 1) { if (!sender.scheduleSend(new TelemetryBufferTelemetriesFetcher(generation), transmitBufferTimeoutInSeconds, TimeUnit.SECONDS)) { logger.error("Failed to schedule send of the buffer to network"); telemetries.clear(); } } } } TelemetryBuffer(TelemetriesTransmitter<T> sender, LimitsEnforcer maxTelemetriesInBatchEnforcer, LimitsEnforcer transmitBufferTimeoutInSecondsEnforcer); void setMaxTelemetriesInBatch(int value); int getMaxTelemetriesInBatch(); void setTransmitBufferTimeoutInSeconds(int value); int getTransmitBufferTimeoutInSeconds(); void add(T telemetry); void flush();  }