Send CANCEL frame on stream request cancelled by client

Author: drstranges

lib/core/rsocket_requester.dart

@@ -45,7 +45,10 @@ class CompleterSubscriber implements Subscriber {
 }
 
 class StreamSubscriber implements Subscriber {
- StreamController controller = StreamController();
+ final StreamController controller;
+
+ StreamSubscriber({FutureOr<void> onCancel() = null})
+ : controller = StreamController(onCancel: onCancel);
 
  @override
  void onNext(Payload value) {
@@ -123,7 +126,10 @@ class RSocketRequester extends RSocket {
  var streamId = streamIdSupplier.nextStreamId(senders);
  connection.write(FrameCodec.encodeRequestStreamFrame(
  streamId, MAX_REQUEST_N_SIZE, payload));
- var streamSubscriber = StreamSubscriber();
+ var streamSubscriber = StreamSubscriber(onCancel: () {
+ connection.write(FrameCodec.encodeCancelFrame(streamId));
+ senders.remove(streamId);
+ });
  senders[streamId] = streamSubscriber;
  return streamSubscriber.payloadStream();
  };