Allow user to not use all the streamed results

Previously, if you didn't use all the returned results from a query
(for example, if you called results.first), then you'd subsequently
get an error. Now it correctly handles this case.

Fixes #43

lib/src/prepared_statements/execute_query_handler.dart

@@ -17,6 +17,7 @@ class _ExecuteQueryHandler extends _Handler {
   List _preparedValues;
   _OkPacket _okPacket;
   bool _executed;
+  bool _cancelled = false;
 
   _ExecuteQueryHandler(_PreparedQuery this._preparedQuery, bool this._executed, List this._values) {
     _fieldPackets = <_FieldImpl>[];
@@ -290,6 +291,10 @@ class _ExecuteQueryHandler extends _Handler {
 
   _HandlerResponse processResponse(Buffer response) {
     var packet;
+    if (_cancelled) {
+      _streamController.close();
+      return new _HandlerResponse(finished: true);
+    }
     if (_state == STATE_HEADER_PACKET) {
       packet = checkResponse(response);
     }
@@ -326,6 +331,9 @@ class _ExecuteQueryHandler extends _Handler {
   _handleEndOfFields() {
     _state = STATE_ROW_PACKETS;
     _streamController = new StreamController<Row>();
+    _streamController.onCancel = () {
+      _cancelled = true;
+    };
     this._fieldIndex = _createFieldIndex();
     return new _HandlerResponse(result: new _ResultsImpl(null, null, _fieldPackets, stream: _streamController.stream));
   }

lib/src/prepared_statements/prepared_query.dart

@@ -5,7 +5,7 @@ class _PreparedQuery {
   final List<_FieldImpl> parameters;
   final List<_FieldImpl> columns;
   final int statementHandlerId;
-  dynamic cnx; // should be a Connection
+  _Connection cnx;
 
   _PreparedQuery(_PrepareHandler handler) :
       sql = handler.sql,

test/integration/execute_multi.dart

@@ -4,7 +4,7 @@ void runExecuteMultiTests(String user, String password, String db, int port, Str
   ConnectionPool pool;
   group('executeMulti tests:', () {
     test('setup', () {
-      pool = new ConnectionPool(user:user, password:password, db:db, port:port, host:host, max:1);
+      pool = new ConnectionPool(user:user, password:password, db:db, port:port, host:host, max:2);
       return setup(pool, "stream", "create table stream (id integer, name text)",
       "insert into stream (id, name) values (1, 'A'), (2, 'B'), (3, 'C')");
     });
@@ -27,6 +27,17 @@ void runExecuteMultiTests(String user, String password, String db, int port, Str
       expect(resultList[0][1].toString(), equals('C'));
     });
 
+    test('issue 43', () async {
+      var tran = await pool.startTransaction();
+      var query = await tran.prepare("SELECT * FROM stream");
+      var result = await query.execute();
+
+      await result.first;
+
+      await query.close();
+      await tran.rollback();
+    });
+
     test('close connection', () {
       pool.closeConnectionsWhenNotInUse();
     });