สวัสดีครับ ผมตามมาจาก youtube ละครับ ผมลอง v 0.1.6 แล้วผล คือ client สามารับ sub message ได้ครบถ้วนจนถึง massage สุดท้ายเรียบร้อย ไม่ delay แล้ว แต่ผมยังเจอปัญหาอยู่เมื่อ client ได้รับ sub message ติดๆกันหรือพร้อมๆกัน เมื่อ log ดู ใน file client.dart สามารถรับ op case 'msg' ได้ครบถ้วนทุก event และเข้า _processMsg() ทุก event แต่เวลาที่จะส่ง payload เข้า _subs[sid].add (Message... มันเหมือน save ทับ message กันครับ ส่งผลให้ได้ msg เพียงอันสุดท้ายที่ส่งต่อไป snapshot.data

ผมคิดว่าน่าจะแปลง <nats.Message> เป็น list ของ messages น่าจะแก้ปัญหานี้ได้ แล้วเมื่อเวลาจะใช้ snapshot.data ผมก็จะใช้ท่า forEach((data) {}) เอาแต่ละ message ไปใช้งานได้

แต่...ผมไม่มีความสามารถแก้ไขได้สำเร็จ คงต้องรบกวนท่านอีกครั้งแล้วหล่ะครับ ><"

ผมเจอเคสที่ publish ไปแล้วไม่รู้เลยว่า subscribe ได้รับ msg หรือเปล่า ผมจะเอา [x] - Request, Respond ตัวนี้มาใช้ได้ไหมครับ ถ้าได้รบกวนท่านอาจารย์ด้วยนะครับ

When there are errors encountered when connecting to a ws address. The connect goes through successfully and there's no way to hook a handler for when the code into the onError section in the connect implementation below:

try { _channel = WebSocketChannel.connect(uri); status = Status.connected; _connectCompleter.complete();

      _addConnectOption(_connectOption);
      _backendSubscriptAll();
      _flushPubBuffer();

      _buffer = [];
      _channel!.stream.listen((d) {
        // _socket!.listen((d) {
        _buffer.addAll(d);
        while (
            _receiveState == _ReceiveState.idle && _buffer.contains(13)) {
          _processOp();
        }
      }, onDone: () {
        status = Status.disconnected;
        _channel!.sink.close();
        // _socket!.close();
      }, onError: (err) {
        status = Status.disconnected;
        _channel!.sink.close();
        // _socket!.close();
      });
      return;
    } catch (err) {
      close();
      _connectCompleter.completeError(err);
    }
    
    Can be easily reproduced using the example code
natsClient = nats.Client();
natsClient.connect(Uri.parse('ws://localhost:1234');
fooSub = natsClient.sub('foo');
barSub = natsClient.sub('bar');

You don't get an error even though 1234 is an invalid port and it goes all the way past the sub without getting any indication that something has gone wrong

1. nodejs pub ไป -> (100%)nats-server(100%) -> flutter(ทั้ง ios+android) sub จะเข้าทุก event ก็ต่อเมื่อ แต่ละ pub ที่ส่งจาก nats-server มีช่วงห่างระหว่างเวลาประมาณตั้งแต่ 300ms ขึ้นไป แต่หาก มี pub จาก client อื่นๆ หรือจาก nodejs ตัวเดียวกันส่งทับซ้อนเข้ามาพร้อมๆกันในช่วงเวลาไม่เกิน 300ms จะทำให้ flutter(ทั้ง ios+android) สามารถรับ sub ได้แค่ event แรก event เดียวเท่านั้น ส่วน event ที่เหลือและที่เพิ่มเข้ามาหลังจากนั้น flutter(ทั้ง ios+android) จะรับ sub ได้เพียงครั้งละ 1 event (ดูจาก log ของ snapshot.data ที่รับจาก sub stream) แต่ปรากฎว่า event ที่ยังไม่ได้รับนั้นจะถูกส่งมาตามลำดับเรื่อยๆ ทีละ event ข้อมูลไม่สูญหายด้วย(ตรงนี้ผมหาคำตอบไม่เจอ) ส่งผลให้การสื่อสาร delay ไปหลายๆ event ตามจำนวนที่รับ sub พร้อมๆกัน สำหรับตัวแปรอื่นๆ ผมได้ทดสอบไปหมดแล้วผ่านหมดตามสถานการณ์ข้อ 1-4 ข้างบน

I was having problems with subscription messages with larger payloads (like 20000 bytes) not working properly. It looks like the code assumes that if a CR is present in the buffer, then there is a full protocol message in the buffer ... but for larger payloads on the MSG type message, the full payload can take longer to arrive.

I am using the code below to get around this, seems to work ok. Basically it does a check to see if the buffer contains a MSG, and if so plucks out the length of the message, then checks to see if the buffer contains the whole payload ... and only calls the processOp() if the payload is complete ... otherwise it breaks out of the while() loop

https://github.com/chartchuo/dart-nats/blob/3bfd99244614110803d62fe38fdd1bdf33537fae/lib/src/client.dart#L118

            while (_receiveState == _ReceiveState.idle && _buffer.contains(13)) {

              var n13 = _buffer.indexOf(13);
              var msgFull = String.fromCharCodes(_buffer.take(n13)).toLowerCase().trim();
              var msgList = msgFull.split(' ');
              var msgType = msgList[0];
              //print('... process $msgType ${_buffer.length}');

              if (msgType == 'msg') {
                var len = int.parse((msgList.length == 4 ? msgList[3] : msgList[4]));
                if (len > 0 && _buffer.length < (msgFull.length + len + 4)) {
                  break; // not a full payload, go around again
                }
              }

              _processOp();
            }

for a larger MSG payload, the commented out print statement produces ...

... process msg 17
... process msg 4113
... process msg 8209
... process msg 12305
... process msg 16401
... process msg 20497
... process msg 21079

When connect to nats://demo.nats.io in macOS and mobile, I have this error

[ERROR:flutter/runtime/dart_vm_initializer.cc(41)] Unhandled Exception: RangeError (start): Invalid value: Only valid value is 0: 2
#0      RangeError.checkValidRange (dart:core/errors.dart:358:7)
#1      _TypedIntListMixin.sublist (dart:typed_data-patch/typed_data_patch.dart:463:31)
#2      Client._sign (package:nats_test/src/client.dart:118:21)
#3      Client._processOp (package:nats_test/src/client.dart:414:15)
#4      Client._steamHandle.<anonymous closure> (package:nats_test/src/client.dart:158:9)
#5      _RootZone.runUnaryGuarded (dart:async/zone.dart:1586:10)
#6      _BufferingStreamSubscription._sendData (dart:async/stream_impl.dart:339:11)
#7      _DelayedData.perform (dart:async/stream_impl.dart:515:14)
#8      _PendingEvents.handleNext (dart:async/stream_impl.dart:620:11)
#9      _PendingEvents.schedule.<anonymous closure> (dart:async/stream_impl.dart:591:7)
#10     _microtaskLoop (dart:async/schedule_microtask.dart:40:21)
#11     _startMicrotaskLoop (dart:async/schedule_microtask.dart:49:5)

After that, I try to comment on this code await _sign(); The connection will be connected

I have an error Unsupported operation: Platform._version, when connecting to wss://{domain}:{port} with my Chrome browser.

I try WebSocket change to

import 'package:web_socket_channel/web_socket_channel.dart';
 WebSocketChannel.connect(uri);

The connection will be connected.

WebSocket from 'dart:html' also works well for the web.

While reading the documents it says that Jetstream is built in yet I don't see any mechanic that other libs have IE jetstream() jetstreamManager(). Did I miss something?

Hi,

migrating from 2.0 to 3.0, I am unable to log with user/password.

flutter snippet :

var natsClient = Nats.Client();
var connectOptions = Nats.ConnectOption(
            user: natsUser,
            name: natsUser,
            pass: natsPassword,
            tlsRequired: false,
          );
          Uri uri = Uri.parse("ws://$natsURL:4222");
          natsClient.connect(uri, connectOption: connectOptions, retryInterval: 5, timeout: 10, retry: true);

Where natsUser is actually equal to "user".

leads to failure to connect. On the server side :

[9] 2021/06/23 11:24:17.737836 [DBG] 10.2.1.90:36448 - cid:616 - Client connection created
[9] 2021/06/23 11:24:17.905280 [ERR] 10.2.1.90:36448 - cid:616 - authentication error - User ""
[9] 2021/06/23 11:24:17.905745 [TRC] 10.2.1.90:36448 - cid:616 - ->> [-ERR Authorization Violation]
[9] 2021/06/23 11:24:17.905969 [DBG] 10.2.1.90:36448 - cid:616 - Client connection closed: Authentication Failure

With 2.0, I could log in without problem.

Thank you !

In current code, if hostname is wrong, socket.connect throw exception, but await for global client.connect connect will be in infinite loop. Fix is with add error setting at exception caught into _connectCompleter. After that global await will received Future object with error.

Problem found on flutter app

is there any example if we use MobX with Observable instead of streamBuilder?

I am updating this thread because I cannot listen anything when subscribing my nats server. it not works. I have following your sample code, and success to connect. Honestly, the message whom published from server did not appear..

repoduce step

     var client = Client();
      var gotit = false;
      client.statusStream.listen(
        (s) {
          print(s);
        },
        onError: (e) {
          gotit = true;
        },
      );
      try {
        await client.connect(Uri.parse('ws://localhost:1234'),
            retry: false, retryInterval: 1);
      } on NatsException {
        gotit = true;
      } on WebSocketChannelException {
        gotit = true;
      } catch (e) {
        gotit = true;
      }