是否可以将ZMQ代理用作“交换机”？

Question

I'm going to implement the following diagram: ZMQ-proxy

Just to let all "data generators" publish data to all "data receivers". Data receivers should still have ability to set subscritions.

// Proxy
package main

import zmq "github.com/pebbe/zmq4"

func main() {
    publisher, _ := zmq.NewSocket(zmq.XPUB)
    defer publisher.Close()
    publisher.Bind("tcp://*:8080")

    subscriber, _ := zmq.NewSocket(zmq.XSUB)
    defer subscriber.Close()
    subscriber.Bind("tcp://*:8081")

    zmq.Proxy(subscriber, publisher, nil)
}

And "data generators":

// Data generator
package main

import (
    zmq "github.com/pebbe/zmq4"
    "time"
)

func main() {
    publisher, _ := zmq.NewSocket(zmq.PUB)
    defer publisher.Close()
    publisher.Connect("tcp://127.0.0.1:8081")

    for {
        publisher.Send("Some msg", 0)
        time.Sleep(time.Second)
    }
}

And "data receivers":

// Data receiver
package main

import (
    "fmt"
    zmq "github.com/pebbe/zmq4"
)

func main() {
    subscriber, _ := zmq.NewSocket(zmq.SUB)
    defer subscriber.Close()
    subscriber.Connect("tcp://127.0.0.1:8080")

    for {
        msg, _ := subscriber.Recv(0)
        fmt.Println(msg)
    }
}

It seems to me that it should work, but it isn't. What am I doing wrong?

Answer 1

To receive any data subscriber socket should specify SetSubscibe with some filter. see Official documentation