mangos_test.go

package main

import (
	"fmt"
	"strconv"
	"time"

	"github.com/gdamore/mangos"
	"github.com/gdamore/mangos/protocol/pub"
	"github.com/gdamore/mangos/protocol/sub"
	"github.com/gdamore/mangos/transport/tcp"
)

func main() {
	pubA, err := pub.NewSocket()
	if err != nil {
		panic(err)
	}
	pubA.AddTransport(tcp.NewTransport())
	if err := pubA.Listen("tcp://:6000"); err != nil {
		panic(err)
	}

	pubB, err := pub.NewSocket()
	if err != nil {
		panic(err)
	}
	pubB.AddTransport(tcp.NewTransport())
	if err := pubB.Listen("tcp://:7000"); err != nil {
		panic(err)
	}

	subA, err := sub.NewSocket()
	if err != nil {
		panic(err)
	}
	subA.AddTransport(tcp.NewTransport())
	subA.Dial("tcp://localhost:7000")
	subA.SetOption(mangos.OptionSubscribe, []byte{})

	subB, err := sub.NewSocket()
	if err != nil {
		panic(err)
	}
	subB.AddTransport(tcp.NewTransport())
	subB.Dial("tcp://localhost:6000")
	subB.SetOption(mangos.OptionSubscribe, []byte{})

	time.Sleep(time.Second)

	go func() {
		go func() {
			for i := 0; i < 100; i++ {
				if err := pubA.Send([]byte("c: " + strconv.Itoa(i))); err != nil {
					panic(err)
				}
				time.Sleep(time.Second)
			}
		}()
		for i := 0; i < 100; i++ {
			if err := pubA.Send([]byte("a: " + strconv.Itoa(i))); err != nil {
				panic(err)
			}
			time.Sleep(time.Second)
		}
	}()

	go func() {
		for i := 0; i < 100; i++ {
			if err := pubB.Send([]byte("b: " + strconv.Itoa(i))); err != nil {
				panic(err)
			}
			time.Sleep(time.Second)
		}
	}()

	go func() {
		for {
			msg, _ := subB.Recv()
			fmt.Println(string(msg))
		}
	}()

	for {
		msg, _ := subA.Recv()
		fmt.Println(string(msg))
	}
}

I've discovered an issue in a project of mine using mangos pub/sub. In it, I have two processes, A and B, each with a publisher and a subscriber. Subscriber A connects to Publisher B and Subscriber B connects to Publisher A.

I've noticed that some published messages are not being received on either side. Many of the publishes on either side occur in different goroutines. The above code is an attempt at reproducing this problem. From what I can tell, it seems to be related to publishing from different goroutines.

Looking at this code, we would expect to see something along the lines of the following (or some variant of it based on how the goroutines are scheduled):

a: 0
c: 0
b: 0
c: 1
a: 1
b: 1
a: 2
c: 2
b: 2
...

However, when you run this code, you'll see that c is never printed. If you comment out the for loop which publishes a, it will be printed.

It looks like it's because the messages are being dropped here:

https://github.com/gdamore/mangos/blob/master/protocol/pub/pub.go#L78

Turns out what's happening is the messages are being dropped by the publisher because it has an unbuffered channel: https://github.com/gdamore/mangos/blob/aee5ac9cca8ad3f72fd5e5792257ca6d4f593d32/protocol/pub/pub.go#L86