From 53ecd7a4c1e7b26aab231d986786884cb873d779 Mon Sep 17 00:00:00 2001
From: Achilleas Anagnostopoulos <achilleas@taxibeat.com>
Date: Fri, 5 Dec 2014 07:17:24 +0200
Subject: [PATCH] Make sure that socket is flushed before adding new data

---
 lib/src/connection/connection.dart | 12 ++++++++----
 1 file changed, 8 insertions(+), 4 deletions(-)

diff --git a/lib/src/connection/connection.dart b/lib/src/connection/connection.dart
index 36a532d..cf68bb1 100644
--- a/lib/src/connection/connection.dart
+++ b/lib/src/connection/connection.dart
@@ -27,6 +27,7 @@ class Connection {
   Map<int, Completer> _pendingResponses;
   Completer _connected;
   Completer _drained;
+  Future _socketFlushed;
 
   Connection(String this.connId, String this.host, int this.port, {PoolConfiguration config, String this.defaultKeyspace}) {
     // If no config is specified, use the default one
@@ -82,6 +83,7 @@ class Connection {
     connectionLogger.info("[${connId}] Trying to connect to ${host}:${port} [attempt ${_connectionAttempt + 1}/${_poolConfig.maxConnectionAttempts}]");
     Socket.connect(host, port).then((Socket s) {
       _socket = s;
+      _socketFlushed = new Future.value(true);
 
       // Initialize our writer pool and set the reservation timeout
       _reservedFrameWriters.clear();
@@ -227,15 +229,17 @@ class Connection {
   Future _writeMessage(RequestMessage message) {
 
     Completer reply = new Completer();
-    // Block till we get back a frame writer
-    //new Future.sync(_frameWriterPool.reserve)
-    _frameWriterPool.reserve()
+    // Make sure we have flushed our socket data and then
+    // block till we get back a frame writer
+    _socketFlushed
+    .then( (_) => _frameWriterPool.reserve())
     .then((FrameWriter writer) {
       _reservedFrameWriters[ writer.getStreamId()] = writer;
       _pendingResponses[ writer.getStreamId() ] = reply;
       connectionLogger.fine("[${connId}] [stream #${writer.getStreamId()}] Sending message of type ${Opcode.nameOf(message.opcode)} (${message.opcode}) ${message}");
       writer.writeMessage(message, _socket, compression : _poolConfig.compression);
-      return _socket.flush();
+      _socketFlushed = _socket.flush();
+      return _socketFlushed;
     })
     .catchError((e, trace) {
       // Wrap SocketExceptions