MessageBatchingWriter { public void write(FudgeMsg message) { Queue<FudgeMsg> messages = null; synchronized (this) { if (_writingThreadActive) { if (_messages != null) { _messages.add(message); return; } else { messages = new LinkedList<FudgeMsg>(); _messages = messages; } } else { _writingThreadActive = true; } } boolean waitForOtherThreads = false; if (messages == null) { try { beforeWrite(); getFudgeMsgWriter().writeMessage(message); } finally { synchronized (this) { if (_messages != null) { _writingLock.release(); } else { _writingThreadActive = false; if (_nanoFlushDelay > 0) { if (!_flushRequired) { waitForOtherThreads = true; _flushRequired = true; } } else { final FudgeMsgWriter writer = getFudgeMsgWriter(); if (writer != null) { writer.flush(); } } } } } } else { try { _writingLock.acquire(); } catch (InterruptedException e) { throw new OpenGammaRuntimeException("Interrupted", e); } synchronized (this) { _messages = null; } try { beforeWrite(); do { getFudgeMsgWriter().writeMessage(message); message = messages.poll(); } while (message != null); } finally { synchronized (this) { if (_messages != null) { _writingLock.release(); } else { _writingThreadActive = false; if (_nanoFlushDelay > 0) { if (!_flushRequired) { waitForOtherThreads = true; _flushRequired = true; } } else { getFudgeMsgWriter().flush(); } } } } } if (waitForOtherThreads) { LockSupport.parkNanos(_nanoFlushDelay); synchronized (this) { if (_flushRequired) { if (!_writingThreadActive) { getFudgeMsgWriter().flush(); } _flushRequired = false; } } } } MessageBatchingWriter();  MessageBatchingWriter(final FudgeMsgWriter out);  MessageBatchingWriter(final FudgeContext fudgeContext, final OutputStream out); void setFlushDelay(final int microseconds); void setFudgeMsgWriter(final FudgeMsgWriter out); void setFudgeMsgWriter(final FudgeContext fudgeContext, final OutputStream out); void write(FudgeMsg message);  }
@Test(invocationCount = 5, successPercentage = 19) public void concurrentWritesWithBatching() throws InterruptedException { final Thread[] threads = new Thread[6]; for (int i = 0; i < threads.length; i++) { final int id = i; threads[i] = new Thread() { @Override public void run() { try { Thread.sleep(10); } catch (InterruptedException e) { fail(); } switch (id) { case 0: case 1: case 3: case 4: threads[id + 1].start(); break; } _writer.write(FudgeContext.EMPTY_MESSAGE); switch (id) { case 0: threads[3].start(); break; } } }; } threads[0].start(); for (int i = 0; i < threads.length; i++) { threads[i].join(); } assertEquals(3, _out._writes.size()); assertNotNull(_out._writes.get(threads[0])); assertEquals(FudgeSize.calculateMessageEnvelopeSize(FudgeContext.EMPTY_MESSAGE), _out._writes.get(threads[0]).intValue()); assertNotNull(_out._writes.get(threads[1])); assertEquals(2 * FudgeSize.calculateMessageEnvelopeSize(FudgeContext.EMPTY_MESSAGE), _out._writes.get(threads[1]).intValue()); assertNull(_out._writes.get(threads[2])); assertNotNull(_out._writes.get(threads[3])); assertEquals(3 * FudgeSize.calculateMessageEnvelopeSize(FudgeContext.EMPTY_MESSAGE), _out._writes.get(threads[3]).intValue()); assertNull(_out._writes.get(threads[4])); assertNull(_out._writes.get(threads[5])); }