rsocket-core - npm Package Compare versions

Comparing version 0.0.23 to 0.0.24

build/LiteBuffer.js

@@ -436,3 +436,11 @@ 'use strict';
   constructor(value, byteOffset, length) {
-    super(value, byteOffset, length);
+    if (typeof value == 'number') {
+      super(value);
+    } else {
+      const offset = byteOffset || 0;
+      const realLength =
+        //$FlowFixMe
+        length || (isInstance(value, Array) ? value.length : value.byteLength);
+      super(value, offset, realLength);
+    }
   }
@@ -439,0 +447,0 @@ /**

build/RSocketBinaryFraming.js

@@ -136,2 +136,4 @@ /** Copyright (c) Facebook, Inc. and its affiliates.
       return deserializeRequestChannelFrame(buffer, streamId, flags, encoders);
+    case _RSocketFrame.FRAME_TYPES.METADATA_PUSH:
+      return deserializeMetadataPushFrame(buffer, streamId, flags, encoders);
     case _RSocketFrame.FRAME_TYPES.REQUEST_N:
@@ -176,2 +178,4 @@ return deserializeRequestNFrame(buffer, streamId, flags, encoders);
       return serializeRequestManyFrame(frame, encoders);
+    case _RSocketFrame.FRAME_TYPES.METADATA_PUSH:
+      return serializeMetadataPushFrame(frame, encoders);
     case _RSocketFrame.FRAME_TYPES.REQUEST_N:
@@ -215,2 +219,4 @@ return serializeRequestNFrame(frame, encoders);
       return sizeOfRequestManyFrame(frame, encoders);
+    case _RSocketFrame.FRAME_TYPES.METADATA_PUSH:
+      return sizeOfMetadataPushFrame(frame, encoders);
     case _RSocketFrame.FRAME_TYPES.REQUEST_N:
@@ -643,2 +649,30 @@ return sizeOfRequestNFrame(frame, encoders);
 
+/**
+ * Writes a METADATA_PUSH frame to a new buffer and returns
+ * it.
+ */
+function serializeMetadataPushFrame(frame, encoders) {
+  const metadata = frame.metadata;
+  if (metadata != null) {
+    const buffer = (0, _RSocketBufferUtils.createBuffer)(
+      FRAME_HEADER_SIZE + encoders.metadata.byteLength(metadata)
+    );
+
+    const offset = writeHeader(frame, buffer);
+    encoders.metadata.encode(metadata, buffer, offset, buffer.length);
+    return buffer;
+  } else {
+    const buffer = (0, _RSocketBufferUtils.createBuffer)(FRAME_HEADER_SIZE);
+    writeHeader(frame, buffer);
+    return buffer;
+  }
+}
+
+function sizeOfMetadataPushFrame(frame, encoders) {
+  return (
+    FRAME_HEADER_SIZE +
+    (frame.metadata != null ? encoders.metadata.byteLength(frame.metadata) : 0)
+  );
+}
+
 function deserializeRequestFnfFrame(buffer, streamId, flags, encoders) {
@@ -684,2 +718,21 @@ (0, _invariant.default)(
 
+function deserializeMetadataPushFrame(buffer, streamId, flags, encoders) {
+  (0, _invariant.default)(
+    streamId === 0,
+    'RSocketBinaryFraming: Invalid METADATA_PUSH frame, expected stream id to be 0.'
+  );
+
+  const length = buffer.length;
+  return {
+    flags,
+    length,
+    metadata:
+      length === FRAME_HEADER_SIZE
+        ? null
+        : encoders.metadata.decode(buffer, FRAME_HEADER_SIZE, length),
+    streamId,
+    type: _RSocketFrame.FRAME_TYPES.METADATA_PUSH,
+  };
+}
+
 /**
@@ -686,0 +739,0 @@ * Writes a REQUEST_STREAM or REQUEST_CHANNEL frame to a new buffer and returns

build/RSocketMachine.js

@@ -538,4 +538,14 @@ /** Copyright (c) Facebook, Inc. and its affiliates.
   metadataPush(payload) {
-    // TODO #18065331: implement metadataPush
-    throw new Error('metadataPush() is not implemented');
+    return new _rsocketFlowable.Single((subscriber) => {
+      const metadata = this._serializers.metadata.serialize(payload.metadata);
+      const frame = {
+        flags: 0,
+        metadata,
+        streamId: 0,
+        type: _RSocketFrame.FRAME_TYPES.METADATA_PUSH,
+      };
+      this._connection.sendOne(frame);
+      subscriber.onSubscribe(() => {});
+      subscriber.onComplete();
+    });
   }
@@ -631,2 +641,4 @@ _getNextStreamId(streamIds) {
       case _RSocketFrame.FRAME_TYPES.METADATA_PUSH:
+        this._handleMetadataPush(frame);
+        break;
       case _RSocketFrame.FRAME_TYPES.REQUEST_CHANNEL:
@@ -837,2 +849,11 @@ case _RSocketFrame.FRAME_TYPES.REQUEST_FNF:
 
+  _handleMetadataPush(frame) {
+    const payload = this._deserializeMetadataPushPayload(frame);
+    this._requestHandler.metadataPush(payload).subscribe({
+      onComplete: () => {},
+      onError: (error) => {},
+      onSubscribe: (cancel) => {},
+    });
+  }
+
   _sendStreamComplete(streamId) {
@@ -858,2 +879,5 @@ this._subscriptions.delete(streamId);
     });
+
+    const error = new Error(`terminated from the requester: ${errorMessage}`);
+    this._handleStreamError(streamId, error);
   }
@@ -883,2 +907,6 @@
 
+  _deserializeMetadataPushPayload(frame) {
+    return deserializeMetadataPushPayload(this._serializers, frame);
+  }
+
   /**
@@ -902,1 +930,8 @@ * Handle an error specific to a stream.
 }
+
+function deserializeMetadataPushPayload(serializers, frame) {
+  return {
+    data: null,
+    metadata: serializers.metadata.deserialize(frame.metadata),
+  };
+}

build/RSocketServer.js

@@ -152,2 +152,14 @@ /** Copyright (c) Facebook, Inc. and its affiliates.
                     this._connections.add(serverMachine);
+                    connection.connectionStatus().subscribe({
+                      onNext: (status) => {
+                        if (
+                          status.kind === 'CLOSED' ||
+                          status.kind === 'ERROR'
+                        ) {
+                          this._connections.delete(serverMachine);
+                        }
+                      },
+                      onSubscribe: (subscription) =>
+                        subscription.request(Number.MAX_SAFE_INTEGER),
+                    });
                   } catch (error) {
@@ -165,5 +177,2 @@ connection.sendOne({
                   }
-
-                  // TODO(blom): We should subscribe to connection status
-                  // so we can remove the connection when it goes away
                   break;
@@ -170,0 +179,0 @@ default:

package.json

 {
   "name": "rsocket-core",
   "description": "RSocket core",
-  "version": "0.0.23",
+  "version": "0.0.24",
   "repository": {
@@ -12,7 +12,7 @@ "type": "git",
   "dependencies": {
-    "fbjs": "^2.0.0",
-    "rsocket-flowable": "^0.0.23",
-    "rsocket-types": "^0.0.23"
+    "fbjs": "^3.0.0",
+    "rsocket-flowable": "^0.0.24",
+    "rsocket-types": "^0.0.24"
   },
-  "gitHead": "8e357291ed8d6741460b648dcc96d50aec427d4e"
+  "gitHead": "b279d675698163447a31a920ffd7c50c07d647a6"
 }

New alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Fixed alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Improved metrics

Total package byte prevSize

586954

increased by1.69%

Lines of code

12030

increased by1.65%

Dependency changes

• + Addedfbjs@3.0.5(transitive)

• + Addedrsocket-flowable@0.0.24(transitive)

• + Addedrsocket-types@0.0.24(transitive)

• + Addedua-parser-js@1.0.39(transitive)

• - Removedcore-js@3.39.0(transitive)

• - Removedfbjs@2.0.0(transitive)

• - Removedrsocket-flowable@0.0.23(transitive)

• - Removedrsocket-types@0.0.23(transitive)

• - Removedua-parser-js@0.7.39(transitive)

• Updatedfbjs@^3.0.0

• Updatedrsocket-flowable@^0.0.24

• Updatedrsocket-types@^0.0.24