faire une demande auprès d'un groupe Nats

Question

Voici mon exemple de sub.go :

package main

import (
    "fmt"
    nats "github.com/nats-io/nats.go"
)

type Message struct {
    Status string `json:"Status"`
    Msg    string `json:"Msg"`
}

type Response struct {
    Status string `json:"Status"`
    Msg    string `json:"Msg"`
}

var nc *nats.Conn
var c *nats.EncodedConn

func start(){
    nc, _ := nats.Connect("127.0.0.1:4222")
    c, _ := nats.NewEncodedConn(nc, nats.JSON_ENCODER)
    c.QueueSubscribe("subject_toto", "queue_titi", func(_, reply string, message  *Message) {
        fmt.Printf("%+v\n", message)
        var response Response
        response.Msg = "message received"
        response.Status = "Ok"
        c.Publish("reply", response)
        c.Flush()
    })
    c.Flush()
}

func main() {
    fmt.Println("begin")
    go start()
    defer c.Close()
    fmt.Scanln()
    nc.Drain()
    // Close connection
    nc.Close()
    fmt.Println("done")
}

Et cela fonctionne parfaitement, alors maintenant je voudrais publier un message dans cette file d'attente, voici mon pub.go :

package main

import (
    "fmt"
    nats "github.com/nats-io/nats.go"
    "time"
)

type Message struct {
    Status string `json:"Status"`
    Msg    string `json:"Msg"`
}

type Response struct {
    Status string `json:"Status"`
    Msg    string `json:"Msg"`
}

var nc *nats.Conn
var c *nats.EncodedConn

func start(){
    var err error
    var message Message
    var response Response
    nc, _ := nats.Connect("127.0.0.1:4222")
    c, _ := nats.NewEncodedConn(nc, nats.JSON_ENCODER)
    message.Status = "Ok"
    message.Msg = "hello"
    err = c.Request("subject_toto", message, &response, 6*time.Second)
    if err != nil {
        fmt.Printf("%+v\n", err)
    }
    fmt.Printf("%+v\n","response")
    fmt.Printf("%+v\n", response)
    defer c.Close()
}

func main() {
    fmt.Println("begin")
    start()
    fmt.Println("done")
}

Mais lorsque j'essaie de le publier, je reçois une réponse vide :

response {Status : Msg:}

Un début de réponse semble utiliser PublishRequest mais il semble que je ne puisse envoyer qu'une chaîne de caractères au serveur et non une structure.

Answer 1

Le problème se situe au niveau de cette ligne :

c.Publish("reply", response)

Lorsque vous envoyez une demande et que votre abonné va y répondre, il doit y avoir un "sujet de réponse" appelé boîte de réception. Cette boîte de réception est définie dans la section reply arg de la fonction de traitement.

Vous devez donc publier votre réponse dans le sujet renvoyé par la fonction reply de la fonction de traitement dans QueueSubscribe() Il faut donc remplacer cette ligne par :

c.Publish(reply, response)

La valeur de la réponse arg est importante pour la communication et se présente comme suit : _INBOX.bw5EtJShBTI9OQdvxFOBlz.VxsGBcjH

Answer 2

Voici la version fonctionnelle de sub :

package main

import (
    "encoding/json"
    "fmt"
    nats "github.com/nats-io/nats.go"
)
type Message struct {
    Status string `json:"Status"`
    Msg    string `json:"Msg"`
}

type Response struct {
    Status string `json:"Status"`
    Msg    string `json:"Msg"`
}

var nc *nats.Conn
var c *nats.EncodedConn

func start(){

    nc, _ := nats.Connect("127.0.0.1:4222")
    c, _ := nats.NewEncodedConn(nc, nats.JSON_ENCODER)
    c.QueueSubscribe("subject_toto", "queue_titi",func(msg *nats.Msg) {
        var message Message
        err := json.Unmarshal([]byte(msg.Data), &message)
        if err != nil {
            fmt.Printf("%+v\n", err)
        }
        fmt.Printf("%+v\n", "message from pub")
        fmt.Printf("%+v\n", message)
        var response Response
        response.Msg = "message received"
        response.Status = "Ok"
        fmt.Printf("%+v\n", "response to  sub")
        fmt.Printf("%+v\n", response)
        byteConfApi, err2 := json.Marshal(response)
        if err2 != nil {
            fmt.Printf("%+v\n", err2)
        }
        msg.Respond(byteConfApi)
        c.Flush()
    })
}

func main(){
    fmt.Println("begin")
    go  start()
    defer c.Close()
    fmt.Scanln()
    nc.Drain()
    // Close connection
    nc.Close()
    fmt.Println("done")
}