IpcSenderService extends MonitoredProcess { synchronized public void add(InternalEvent ievent) throws TransportException { if (this.hasUnrecoverableException.get()) { throw new TransportException("TransportThread was unsuccessful when adding an event."); } LinkedHashMap<String, String> partitions = ievent.getPartitions(); synchronized (buffers) { if (partitions == null) { partitions = new LinkedHashMap<String, String>(0); } if (!this.buffers.containsKey(partitions)) { partitions = new LinkedHashMap<String, String>(partitions); this.buffers.put(partitions, this.transportFactory.newTransportBuffer()); } TransportBuffer buffer = this.buffers.get(partitions); try { buffer.add(ievent); return; } catch (IllegalStateException e) { send(buffer, partitions); } catch (IOException e) { throw new TransportException("Exception occurred while adding to buffer", e); } this.buffers.remove(partitions); buffer = this.transportFactory.newTransportBuffer(); partitions = new LinkedHashMap<String, String>(partitions); this.buffers.put(partitions, buffer); try { buffer.add(ievent); } catch (IllegalStateException e) { throw new TransportException("Buffer is full despite being newly created", e); } catch (IOException e) { throw new TransportException("Exception ocurred while adding to buffer", e); } } } IpcSenderService(TransportFactory factory); synchronized void add(InternalEvent ievent); synchronized void flush(); synchronized void shutdown(); boolean hasUnrecoverableException(); void setHasUnrecoverableException(boolean has); TransportFactory getTransportFactory(); void setTransportFactory(TransportFactory transportFactory); Context getContext(); void setContext(Context context); final Map<LinkedHashMap<String, String>, TransportBuffer> buffers; }
@Test(expected = TransportException.class) public void testThreadExceptionDuringAdd() throws TransportException, InterruptedException { DummyTransporter mockDummyTransporter = mock(DummyTransporter.class); DummyTransporterFactory tfactory = new DummyTransporterFactory(); tfactory.transporter = mockDummyTransporter; doThrow(new TransportException("expected exception in test")).when(mockDummyTransporter) .sendBatch(any(DummyTransportBuffer.class)); IpcSenderService ipc = new IpcSenderService(tfactory); for (int i = 0; i < 5; i++) { ipc.add(mock(InternalEvent.class)); } ipc.add(mock(InternalEvent.class)); Thread.sleep(5); ipc.add(mock(InternalEvent.class)); }