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