Merge pull request #7 from drstranges/master

Send CANCEL frame on request-stream cancelled by client

Author: linux-china

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();
     };

lib/frame/frame.dart

@@ -472,6 +472,16 @@ class FrameCodec {
     refillFrameLength(frameBuffer);
     return frameBuffer.toUint8Array();
   }
+
+  static Uint8List encodeCancelFrame(int streamId) {
+    var frameBuffer = RSocketByteBuffer();
+    frameBuffer.writeI24(0); // frame length
+    frameBuffer.writeI32(streamId); //stream id
+    frameBuffer.writeI8(frame_types.CANCEL << 2);
+    frameBuffer.writeI8(0);
+    refillFrameLength(frameBuffer);
+    return frameBuffer.toUint8Array();
+  }
 }
 
 Payload decodePayload(

pubspec.yaml

@@ -7,6 +7,6 @@ issue_tracker: https://github.com/rsocket/rsocket-dart/issues
 environment:
   sdk: '>=2.7.0 <3.0.0'
 dependencies:
-  rxdart: ^0.27.0
+  rxdart: ^0.27.1
 dev_dependencies:
   test: ^1.6.0