Protobufjs Codec and general EncConn support

[delaneyj]

Not ready for a PR but more of a discussion. Wanted to match the protobuf support and EncodedConn similar to the Go API using the pbjs/pbts tooling from protobufjs.

So I mocked out an encoder.ts

import { Codec, Msg, NatsConnection, NatsError, Subscription } from 'nats.ws'
import { Reader, Writer } from 'protobufjs'

type suscribeCallback = (err: NatsError | null, msg: Msg) => void

export class EncoderConn<T> {
  constructor(readonly conn: NatsConnection, readonly enc: Codec<T>) {}

  // Publish publishes the data argument to the given subject. The data argument
  // will be encoded using the associated encoder.
  publish(subject: string, v: any) {
    const b = this.enc.encode(v)
    this.conn.publish(subject, b)
  }

  // PublishRequest will perform a Publish() expecting a response on the
  // reply subject. Use Request() for automatically waiting for a response
  // inline.
  publishRequest(subject: string, reply: string, v: any) {
    const b = this.enc.encode(v)
    this.conn.publish(subject, b, {
      reply,
    })
  }

  subscribe(subject: string, callback: suscribeCallback): Subscription {
    return this.conn.subscribe(subject, {
      callback,
    })
  }

  queueSubscribe(
    subject: string,
    queue: string,
    callback: suscribeCallback,
  ): Subscription {
    return this.conn.subscribe(subject, {
      queue,
      callback,
    })
  }
}

export interface Encoder {
  encode(message: any, writer?: Writer): Writer
  decode(reader: Reader | Uint8Array, length?: number): any
}

export function PbjsCodec<T extends Encoder>(t: T): Codec<any> {
  return {
    encode(d: any): Uint8Array {
      debugger
      return t.encode(d).finish()
    },
    decode(a: Uint8Array): any {
      debugger
      return t.decode(a)
    },
  }
}

and tested with

import { connect } from 'nats.ws'

import { pb } from '../gen/pb'
import { EncoderConn, PbjsCodec } from './codec'

const { Book } = pb

export async function test() {
  const conn = await connect({
    servers: 'ws://localhost:9090',
    user: 'foo',
    pass: 'arstars',
  })
  const ec = new EncoderConn(conn, PbjsCodec(Book))

  const b = Book.create({
    isbn: Math.floor(Math.random() * Number.MAX_SAFE_INTEGER),
  })
  ec.publish('foo', b)
}

So you get full static typing and small sizes on the wire. Still have access to json by doing .toObject(). Just looking to see if this sparks any ideas from anyone else.

[aricart]

@delaneyj for the deno based clients, the correct thing is to create a new wrapper client. Currently exploring this for JetStream, and want to offer a similar pattern. This will allow all clients that use the same platform to have same functionality.

[gedw99]

Hey @aricart and @delaneyj

Is there a sample using protobuf ?? Would be really awesome to have the Schema Evolution...

[delaneyj]

Hey @aricart and @delaneyj

Is there a sample using protobuf ?? Would be really awesome to have the Schema Evolution...

No, I ended up just doing code generation to make this happen

[aricart]

The main thing to think about is that adding 3rd party dependencies for out of the box encodings is problematic. Think about in other JavaScript runtimes, even TextEncoder is not available.

But having a general facility where the subscription callback or iterator receive messages in a decoded format, definitely has merit.

So internally there's a class that can do encoded subscriptions. This has not been surfaced up because I believe it is possibly the wrong approach, and want to take another crack at it before 3.0 is out. In the mean while if you want to play on the edge:

https://github.com/nats-io/nats.deno/blob/main/nats-base-client/typedsub.ts#L82

[gedw99]

thanks @aricart makes sense. wil give it a crack !!