I'm working on convert Nats core to Nats jetstream on a go project. I'm using nats.go and as I notice EncodedConn request is used everywhere in transporter files. But EncodedConn doesn't support Jetstream "noticed during debugging". I had to use Nats native request method with Marshal , Unmarshal manually. And I had to repeat this twice, one for subscription and another for transporter. Is there anyway to make it simplest ? I shared a code snippet to make it clear.
I couldn't be able to use encodedconn cause the return type is different than *nats.Msg and I encountered this error "msgpack: invalid code=7b decoding map length"
Transporter File
func (t *Transport) GetProvidersLastblock(provider string) (lastblock LastblockDTO, err error) {
req := LastblockDTO{Provider: provider,}
listblockJson, _ := json.Marshal(req)
msg ,err := t.Conn.Request("LASTBLOCK.GETPROVIDER", listblockJson, GlobalNatsLastblockTimeout)
json.Unmarshal(msg.Data, &lastblock)
return
}
Subscribe File
func getProvidersLastblock(srv *service.Service) (sb *nats.Subscription, err error) {
js,_ := srv.Tp.Conn.JetStream()
sb, err = js.Subscribe(subjectName,
func(msg *nats.Msg) {
msg.Ack()
var lastblockDto transporter.LastblockDTO
json.Unmarshal(msg.Data, &lastblockDto)
lastblock, err := srv.Db.GetProviderLastblock(lastblockDto.Provider)
if err != nil {
srv.Log2.Error(err, "fn", "getProvidersLastblock")
}
lastblockJSON, _ := json.Marshal(lastblock)
_, err = srv.JS.Publish(subjectName, lastblockJSON)
if err != nil {
srv.Log2.Error(err, "fn", "publishProvidersLastblock")
}
}, nats.Durable("monitor"), nats.ManualAck())
return
}