pubsubfun.go

package main

import (
	"fmt"
	"log"
	"time"

	"github.com/zeromq/goczmq"
)

const (
	numSocks = 1000
)

type node struct {
	id   int
	port int
	pub  *goczmq.Sock
	sub  *goczmq.Sock
}

func main() {
	nodes := make([]*node, 0)

	for i := 0; i < numSocks; i++ {
		var err error
		node := &node{id: i}

		node.pub = goczmq.NewSock(goczmq.Pub)

		node.port, err = node.pub.Bind("tcp://127.0.0.1:*")
		if err != nil {
			panic(err)
		}

		log.Printf("I: node %d bound on port %d", node.id, node.port)

		node.sub = goczmq.NewSock(goczmq.Sub)
		nodes = append(nodes, node)
	}

	for _, subscriber := range nodes {
		for _, publisher := range nodes {
			if subscriber.port != publisher.port {
				subscriber.sub.SetSubscribe("hello")
				err := subscriber.sub.Connect(fmt.Sprintf("tcp://127.0.0.1:%d", publisher.port))
				if err != nil {
					panic(err)
				}
			}
		}
	}

	time.Sleep(10 * time.Millisecond)

	for _, publisher := range nodes {
		err := publisher.pub.SendFrame([]byte(fmt.Sprintf("hello from %d", publisher.id)), 0)
		if err != nil {
			panic(err)
		}
		log.Printf("I: node %d sent hello", publisher.id)
	}

	for _, subscriber := range nodes {
		for i := 0; i < (numSocks - 1); i++ {
			b, _, err := subscriber.sub.RecvFrame()
			if err != nil {
				panic(err)
			}
			log.Printf("I: node %d received %s", subscriber.id, string(b))
		}
	}

	for _, node := range nodes {
		node.pub.Destroy()
		node.sub.Destroy()
	}
}