Uncaught Error: Bad state: StreamSink is bound to a stream

[christophe-blettry]

Hello,
Sometimes (it's not 100% reproducible) I have this error:

Unhandled exception:
Uncaught Error: Bad state: StreamSink is bound to a stream
Stack Trace:
#0      _StreamSinkImpl._controller (dart:io/io_sink.dart:219)
#1      _StreamSinkImpl.add (dart:io/io_sink.dart:143)
#2      _Socket.add (dart:io-patch/socket_patch.dart:1644)
#3      ChunkedOutputWriter.pipe.<anonymous closure> (package:dart_cassandra_cql/driver/stream/chunked_output_writer.dart:53:65)
#4      ListQueue.forEach (dart:collection/queue.dart:404)
#5      ChunkedOutputWriter.pipe (package:dart_cassandra_cql/driver/stream/chunked_output_writer.dart:53:28)
#6      FrameWriter.writeMessage (package:dart_cassandra_cql/driver/protocol/frame/frame_writer.dart:96:29)
#7      Connection._writeMessage.<anonymous closure> (package:dart_cassandra_cql/driver/connection/connection.dart:237:26)
#8      _RootZone.runUnary (dart:async/zone.dart:1155)
#9      _Future._propagateToListeners.handleValueCallback (dart:async/future_impl.dart:484)
#10     _Future._propagateToListeners (dart:async/future_impl.dart:567)
#11     _Future._completeWithValue (dart:async/future_impl.dart:358)
#12     _Future._asyncComplete.<anonymous closure> (dart:async/future_impl.dart:412)
#13     _asyncRunCallbackLoop (dart:async/schedule_microtask.dart:41)
#14     _asyncRunCallback (dart:async/schedule_microtask.dart:48)
#15     _runPendingImmediateCallback (dart:isolate-patch/isolate_patch.dart:84)
#16     _RawReceivePortImpl._handleMessage (dart:isolate-patch/isolate_patch.dart:131)

#0      _rootHandleUncaughtError.<anonymous closure> (dart:async/zone.dart:886)
#1      _asyncRunCallbackLoop (dart:async/schedule_microtask.dart:41)
#2      _asyncRunCallback (dart:async/schedule_microtask.dart:48)
#3      _runPendingImmediateCallback (dart:isolate-patch/isolate_patch.dart:84)
#4      _RawReceivePortImpl._handleMessage (dart:isolate-patch/isolate_patch.dart:131)

My code (Windows 7, Dart 1.8, Cassandra 2.1.2):

====================
main:
====================
void initCassandra(){
  List<String> queries = new List<String>();
   queries.add(common_dao.createSimpleKeyspaceQuery("bi"));
   queries.add(user_dao.getCreateTableQuery());
   cassandra.synchronizedQuery(queries);
}
void main() {
  initCassandra(); 

  HttpServer.bind(InternetAddress.LOOPBACK_IP_V4, 8080).then((server) {
    var router = new Router(server)
          ....
  });
}
========================
cassandra.dart:
========================
library cassandra;

import 'dart:async';
import 'package:dart_cassandra_cql/dart_cassandra_cql.dart' as cql;

void synchronizedQuery(List<String>queries){
  cql.Client client = new cql.Client.fromHostList(['localhost']);
  if(queries.isEmpty){
    return;
  }
  print("synchronizedQuery: query[0]: "+queries[0]);
  Future<cql.ResultMessage> f = client.execute(new cql.Query(queries[0]))
      .catchError((e){print("synchronizedQuery: Error: "+e.toString());});
  int i=1;
  while(i<queries.length){
    String q = queries[i];
    f.then((_){client.execute(new cql.Query(q));});
    i++;
  }
}

Something is it wrong ?
Thanks very much

Christophe

[achilleasa]

Looks like there is a race condition while flushing the data to the socket. Could you try checking out the current master from GH (53ecd7a) and see if it resolves your issue?

[christophe-blettry]

Hello Achilleas,

Thanks for the fixe.
I not able to reproduce the issue.

Christophe