dustin / go-broadcast

A trivial channel pubsub I use in lots of projects.

pubsubbing channels.

This project primarily exists because I've been copying and pasting the exact same two files into numerous projects. It does work well, though.

See the documentation for usage and examples.

go-broadcast's Issues

about Submit method

func (b *broadcaster) Submit(m interface{}) {
if b != nil {
b.input <- m
}
}

why here need to judge b is nil ?

An example issue

when the first worker prints all five by a go routine, the deffer unregister will be called at the end of worker function.
Then the first worker channel unsubscribes, and all five will not be sent to the first worker channel and print.

Block Forever!

I think the broadcast method should use non-block.
If the receiver want to buffer msgs, they should do it themself.

func (b *broadcaster) broadcast(m interface{}) {
	for ch := range b.outputs {
		select {
		case ch <- m:
		default:
		}
	}
}

Or it's the bug now:

func TestBoardcast(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	var wg sync.WaitGroup
	b := broadcast.NewBroadcaster(100)
	for idx := 0; idx < 2; idx++ {
		wg.Add(1)
		go func(idx int) {
			defer wg.Done()

			end := false
			for !end {
				fmt.Printf("waiter%d: Waiting For Target...\n", idx)
				ch := make(chan interface{})
				b.Register(ch)
				fmt.Printf("waiter%d: Registered %v!\n", idx, ch)
				var e interface{}
				select {
				case <-ctx.Done():
					b.Unregister(ch)
					fmt.Printf("waiter%d Done: %v\n", idx, ctx.Err())
					return
				case e = <-ch:
					end = e.(bool)
					fmt.Printf("waiter%d: Receive Target Changed: %t!\n", idx, end)
				}
				b.Unregister(ch)
				fmt.Printf("waiter%d: Unregistered!\n", idx)
			}

			fmt.Printf("waiter%d: Got Target!\n", idx)
		}(idx)
	}

	time.Sleep(1000 * time.Millisecond)
	for idx := 0; idx < 10; idx++ {
		if idx < 5 {
			b.Submit(false)
			fmt.Printf("Sended Target Changed %d: %t!\n", idx, false)
		} else {
			b.Submit(true)
			fmt.Printf("Sended Target Changed %d: %t!\n", idx, true)
		}
	}

	time.Sleep(2 * time.Second)
	cancel()

	wg.Wait()
}

Operation blocks on receive delay

Any registered receiver can block the Broadcaster operation when the receive channel is not read (fast enough). Maybe just a comment would do? I've made another implementation to overcome this issue just now, based on yours.