rsocket / rsocket-dart

Dart implementation of RSocket

RSocket Dart SDK

Examples

Client

import 'package:rsocket/rsocket_connector.dart';
import 'package:rsocket/payload.dart';

void main() async {
  var rsocket = await RSocketConnector.create().connect('tcp://127.0.0.1:42252');
  var result = await rsocket.requestResponse(Payload.fromText('text/plain', 'Ping'));
  print(result.getDataUtf8());
}

Server

import 'package:rsocket/rsocket_server.dart';
import 'package:rsocket/payload.dart';
import 'package:rsocket/rsocket.dart';

void main() async {
  const listenUrl = 'tcp://0.0.0.0:42252';
  var closeable = await RSocketServer.create(requestResponseAcceptor((payload) {
    return Future.value(Payload.fromText('text/plain', 'Pong'));
  })).bind(listenUrl);
  print('RSocket Server started on ${listenUrl}');
}

RSocket Service Proxy & Routing

Please refer https://github.com/rsocket/rsocket-dart/tree/master/lib/route

RSocket

• Operations
  • REQUEST_FNF
  • REQUEST_RESPONSE
  • REQUEST_STREAM
  • REQUEST_CHANNEL
  • METADATA_PUSH
• More Operations
  • Error
  • Cancel
  • Keepalive
• QoS
  • RequestN
  • Lease
• Transport
  • TCP
  • Websocket
• High Level APIs
  • Client
  • Server
• Misc
  • RxDart

References

• RSocket: https://rsocket.io
• Dart: https://dart.dev/

rsocket-dart's Issues

Flutter error with dart:mirrors

I want to use rsocket with my Flutter App.

Expected Behavior

I shall import shelf normally

import 'package:rsocket/shelf.dart';

Actual Behavior

Simply importing this library in a Flutter project causes problem.

import 'dart:mirrors';

So as this library is using it inside route, Flutter crashes.

Steps to Reproduce

To reproduce it, just import the library inside a Flutter project:

import 'package:rsocket/shelf.dart';

Possible Solution

We might use reflectable instead of dart:mirrors as said in the Library here.

Your Environment

rsocket: ^0.1.2
Flutter 2.0.6 • channel stable • https://github.com/flutter/flutter.git
Framework • revision 1d9032c7e1 (3 weeks ago) • 2021-04-29 17:37:58 -0700
Engine • revision 05e680e202
Tools • Dart 2.12.3

Provide TcpDuplexConnection examples

An example for a full duplex connection would be nice

Support Flutter please...

This library does not support Flutter because it uses mirrors.
Please help us use it in Flutter.

Also, update support for the Dart library is very lacking. Since the library version is not specified, the dependent library has been updated, but RSocket does not follow the update, so there is also a version problem.

Web support

Any signs of this plugin working with web?
I get

Error: Unsupported operation: Socket constructor
    at Object.throw_ [as throw] (http://localhost:50672/dart_sdk.js:5080:11)
    at Socket._connect (http://localhost:50672/dart_sdk.js:60222:17)
    at Socket.connect (http://localhost:50672/dart_sdk.js:60205:26)
    at Object.connectRSocket (http://localhost:50672/packages/rsocket/duplex_connection.dart.lib.js:200:36)
    at rsocket_connector.RSocketConnector.create.connect (http://localhost:50672/packages/rsocket/rsocket_connector.dart.lib.js:89:34)
    at connect.next (<anonymous>)
    at runBody (http://localhost:50672/dart_sdk.js:40660:34)
    at Object._async [as async] (http://localhost:50672/dart_sdk.js:40691:7)
    at rsocket_connector.RSocketConnector.create.connect (http://localhost:50672/packages/rsocket/rsocket_connector.dart.lib.js:76:20)
    at Object.addName (http://localhost:50672/packages/dance_partner_finder/api_gateway_rsocket_client.dart.lib.js:40:60)
    at http://localhost:50672/packages/dance_partner_finder/dance_partner_finder_bloc.dart.lib.js:173:212
    at _RootZone.runUnary (http://localhost:50672/dart_sdk.js:40511:59)
    at _FutureListener.then.handleValue (http://localhost:50672/dart_sdk.js:35438:29)
    at handleValueCallback (http://localhost:50672/dart_sdk.js:35999:49)
    at _Future._propagateToListeners (http://localhost:50672/dart_sdk.js:36037:17)
    at [_completeWithValue] (http://localhost:50672/dart_sdk.js:35872:23)
    at async._AsyncCallbackEntry.new.callback (http://localhost:50672/dart_sdk.js:35906:35)
    at Object._microtaskLoop (http://localhost:50672/dart_sdk.js:40778:13)
    at _startMicrotaskLoop (http://localhost:50672/dart_sdk.js:40784:13)
    at http://localhost:50672/dart_sdk.js:36261:9
```
while trying to connect.

How to reconnect to server?

I don't know how reconnect to server when server restart.

重连机制

没有重连机制吗？，在断线，或网络异常的情况下

Incorrect FrameType for handling RequestStream Frame.

Expected Behavior

Should receive message stream from dart client

Actual Behavior

Received error :

type 'RequestFNFFrame' is not a subtype of type 'RequestStreamFrame' in type cast
#0      RSocketRequester.receiveFrame (package:rsocket/core/rsocket_requester.dart:290:40)
#1      RSocketRequester.receiveChunk (package:rsocket/core/rsocket_requester.dart:205:9)

Steps to Reproduce

Dart Client (Responder) :

List<int> list = 'this is client response!'.codeUnits;
Uint8List data = Uint8List.fromList(list);

RSocketConnector connector = RSocketConnector.create().acceptor(requestStreamAcceptor((payload) {
	List<Payload> randomPayloadList = [Payload.from(null, data)];
        return Stream.fromIterable(randomPayloadList);
}));

var rsocket = await connector.connect("ws://localhost:7000/rsocket");

Spring Server (Requester):

 rsocketRequester.route("").data("this is server reqeust!")
                .retrieveFlux(String.class).subscribe((v) -> {
                   System.out.println(v);
             });

Possible Solution

rsocket-dart/lib/frame/frame.dart

Lines 40 to 42 in f31cc6e

	case frame_types.REQUEST_STREAM:
	frame = RequestFNFFrame.fromBuffer(header, byteBuffer);
	break;

should be replaced with :

    case frame_types.REQUEST_STREAM:
      frame = RequestStreamFrame.fromBuffer(header, byteBuffer);  // correct frame
      break;

• RSocket version(s) used: 1.0.0
• java : openjdk 17.0.5
• Spring Boot Server

Any example on Server to Client call using WebSocketRSocketResponder ?

Hello,

I'm struggling to integrate WebSocketRSocketResponder class. Could you please provide an example code on how to receive response from Server for different type of calls ( like fireAndForget, requestResponse ... ) ?
Please provide some sample example if possible.

Thank you for all the hard works. :)

Easier way to create routed payloads

Creating a route meta data requires some boiler plate code. We can avoid it like below:

  Payload.routeAndData(String route, String data){
    var compositeMetadata = CompositeMetadata.fromEntries([
      RoutingMetadata(route, List.empty())
    ]);
    this.metadata = compositeMetadata.toUint8Array();
    this.data = Uint8List.fromList(utf8.encode(data));
  }

I tried creating a PR but didn't have permissions.

how to cancel request stream

Expected Behavior

Actual Behavior

Steps to Reproduce

@Test
public void reproCase() {

}

Possible Solution

Your Environment

• RSocket version(s) used:
• Other relevant libraries versions (eg. netty, ...):
• Platform (eg. JVM version (javar -version) or Node version (node --version)):
• OS and version (eg uname -a):

request-channel

Please add request-channel

When REQUEST_CHANNEL to be supported?

Custom metadata via CompositeMetadata not working.

Hi,

I have created a custom Mimetype. Using Javascript rsocket client is working as expected.

But using the dart package, the metadata is being malformed.

Looks like this method is not working well for not well known mime types.

file: composite_metadata.dart

void addExplicitMimeType(String? mimeType, Uint8List? content) {
if (WellKnownMimeType.isWellKnownType(mimeType)) {
addWellKnownMimeType(
WellKnownMimeType.getMimeTypeId(mimeType)!, content!);
} else {
var mimeTypeArray = utf8.encode(mimeType!);
buffer.writeI8(mimeTypeArray.length);
buffer.writeBytes(mimeTypeArray);
buffer.writeI24(content!.length);
buffer.writeUint8List(content);
}
}

if I send a well known type, it works.

dartfmt all files

all dart files should be formated by dartfmt

Please add request-channel

Please add request-channel

Please provide a example to Connect with Spring-Boot

I try to connect with spring boot currently have the next code:

@MessageMapping("my-balance")
public Flux timestamps() {
return Flux.interval(Duration.ofSeconds(1))
.map(sequence -> LocalTime.now().toString());
}

Improvement Suggestion

How I can connect with this?

sending with rsc it's work
java -jar rsc.jar --debug --stream --route rep-balance tcp://192.168.0.17:7000

is posible provide a example to Connect with Spring-Boot?
Thanks, I love it project!

Null safety support?

Will there be migration to null safety support?

How to handle fireAndForget message from server

I was trying handle message from server
A Server send me message by fireAndForget

Server is Spring boot

Server Code:

@Scheduled(fixedRate = 10000)
    public void push() {
        rSocketRequester
                .route("api.push")
                .data("some push message with id " + RandomUtil.getPositiveInt())
                .send()
                .block();
    }

Does it possible handle messages like this?

• RSocket version(s) used: 1.0.0

AuthMetadata Jwt 请求响应模式没有任何 结果 也没有错误

var metadata = CompositeMetadata.fromEntries([
      AuthMetadata.jwt(_token),
      RoutingMetadata("user.get",[]),
    ]);
    var users = Users.create();
    users.meta = _meta;
    users.id = Int64(_account!.id.toInt());
    var userBuf = users.writeToBuffer();
    var meta = metadata.toUint8Array();
    var compositeMetadata2 = CompositeMetadata.fromU8Array(meta);
    compositeMetadata2.forEach((entry) {
      print('mimeType: ' +
          entry.mimeType! +
          ', content: ' +
          entry.content!.length.toString());
    });
    var payload = Payload.from(meta, userBuf);
    _rSocket!.requestResponse!(payload).then((value) {
      var data = value.data!;
      Users users = Users.fromBuffer(data);
      var user = IMIOUsers().toProto(users);
      _meta.roomId = Int64(user.roomId.toInt());
      _account!.avatar = user.avatar;
      _account!.nickname = user.nickname;
      _account!.account = user.account;
    }).catchError((e) {
      var message = e.toString();
      if (message.contains("401")) {
        _getToken();
      }
    }).onError((error, stackTrace) {
      print(error.toString());
      stackTrace.toString();
    });

我那些地方不正确嘛
@linux-china