class MessageBatchingWriter {

  public void write(FudgeMsg message) {
    Queue<FudgeMsg> messages = null;
    synchronized (this) {
      if (_writingThreadActive) {
        if (_messages != null) {
          // Another thread is already blocked, so tag onto that and return
          _messages.add(message);
          return;
        } else {
          // Another thread is already writing
          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) {
            // Another thread is blocking to be the active thread
            _writingLock.release();
          } else {
            // No other messages have been attempted
            _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) {
        // Stop other threads from queuing onto us
        _messages = null;
      }
      try {
        beforeWrite();
        do {
          getFudgeMsgWriter().writeMessage(message);
          message = messages.poll();
        } while (message != null);
      } finally {
        synchronized (this) {
          if (_messages != null) {
            // Another thread is blocking to be the active thread
            _writingLock.release();
          } else {
            // No other messages have been attempted
            _writingThreadActive = false;
            if (_nanoFlushDelay > 0) {
              if (!_flushRequired) {
                waitForOtherThreads = true;
                _flushRequired = true;
              }
            } else {
              getFudgeMsgWriter().flush();
            }
          }
        }
      }
    }
    // TODO: it would be better if this could be offloaded to another thread so that
    // we don't block the caller and only have one thread doing the park.
    if (waitForOtherThreads) {
      // Can't reliably do a sub-millisecond precision sleep, so use park
      LockSupport.parkNanos(_nanoFlushDelay);
      synchronized (this) {
        if (_flushRequired) {
          if (!_writingThreadActive) {
            // No other threads have become active to write data so flush
            getFudgeMsgWriter().flush();
          }
          _flushRequired = false;
        }
      }
    }
  }

  public  MessageBatchingWriter();
  public  MessageBatchingWriter(final FudgeMsgWriter out);
  public  MessageBatchingWriter(final FudgeContext fudgeContext, final OutputStream out);

}