func TestBroker_MaxConcurrentProcesses(t *testing.T) {
	broker := buildBrokerLimit4(t, t.Name())
	broker.Start(context.TODO())
	broker.Register(context.TODO(), instanceID1)
	broker.Register(context.TODO(), instanceID2)
	broker.Register(context.TODO(), instanceID3)
	broker.Register(context.TODO(), instanceID4)
	assert.Equal(t, 4, len(broker.outbox))
	inbox5, _ := broker.Register(context.TODO(), instanceID5)
	assert.Equal(t, 4, len(broker.outbox))
	serMsg := createMessage(t, instanceID5)
	broker.HandleMessage(context.TODO(), "", serMsg)
	waitForMessages(t, inbox5, instanceID5, 1)
	assert.Nil(t, broker.outbox[instanceID1.Uint32()])
	inbox6, _ := broker.Register(context.TODO(), instanceID6)
	assert.Equal(t, 4, len(broker.outbox))
	assert.Nil(t, broker.outbox[instanceID2.Uint32()])
	serMsg = createMessage(t, instanceID6)
	broker.HandleMessage(context.TODO(), "", serMsg)
	waitForMessages(t, inbox6, instanceID6, 1)
}
test that after registering the maximum number of protocols,
the earliest one gets unregistered in favor of the newest one.
package hare
import (
	"context"
	"sync"
	"testing"
	"time"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	"github.com/spacemeshos/go-spacemesh/codec"
	"github.com/spacemeshos/go-spacemesh/common/types"
	"github.com/spacemeshos/go-spacemesh/log/logtest"
	"github.com/spacemeshos/go-spacemesh/p2p/pubsub"
	"github.com/spacemeshos/go-spacemesh/signing"
)
var (
	instanceID0 = types.NewLayerID(0)
	instanceID1 = types.NewLayerID(1)
	instanceID2 = types.NewLayerID(2)
	instanceID3 = types.NewLayerID(3)
	instanceID4 = types.NewLayerID(4)
	instanceID5 = types.NewLayerID(5)
	instanceID6 = types.NewLayerID(6)
	instanceID7 = types.NewLayerID(7)
)
const reqID = "abracadabra"
func trueFunc(context.Context) bool {
	return true
}
func falseFunc(context.Context) bool {
	return false
}
func mustEncode(tb testing.TB, value interface{}) []byte {
	tb.Helper()
	buf, err := codec.Encode(value)
	require.NoError(tb, err)
	return buf
}
type mockClient struct {
	id types.LayerID
}
type MockStateQuerier struct {
	res bool
	err error
}
func NewMockStateQuerier() MockStateQuerier {
	return MockStateQuerier{true, nil}
}
func (msq MockStateQuerier) IsIdentityActiveOnConsensusView(ctx context.Context, edID string, layer types.LayerID) (bool, error) {
	return msq.res, msq.err
}
func createMessage(tb testing.TB, instanceID types.LayerID) []byte {
	tb.Helper()
	sr := signing.NewEdSigner()
	b := newMessageBuilder()
	msg := b.SetPubKey(sr.PublicKey()).SetInstanceID(instanceID).Sign(sr).Build()
	return mustEncode(tb, msg.Message)
}
func TestBroker_Start(t *testing.T) {
	broker := buildBroker(t, t.Name())
	err := broker.Start(context.TODO())
	assert.Nil(t, err)
	err = broker.Start(context.TODO())
	assert.NotNil(t, err)
	assert.Equal(t, "instance already started", err.Error())
}
func TestBroker_Received(t *testing.T) {
	broker := buildBroker(t, t.Name())
	broker.Start(context.TODO())
	inbox, err := broker.Register(context.TODO(), instanceID1)
	assert.Nil(t, err)
	serMsg := createMessage(t, instanceID1)
	broker.HandleMessage(context.TODO(), "", serMsg)
	waitForMessages(t, inbox, instanceID1, 1)
}
func TestBroker_Priority(t *testing.T) {
	broker := buildBroker(t, t.Name())
	wg := sync.WaitGroup{}
	wg.Add(1)
	wg2 := sync.WaitGroup{}
	wg2.Add(1)
	once := sync.Once{}
	broker.eValidator = &mockEligibilityValidator{validationFn: func(context.Context, *Msg) bool {
		once.Do(wg2.Done)
		wg.Wait()
		return true
	}}
	require.NoError(t, broker.Start(context.TODO()))
	outbox, err := broker.Register(context.TODO(), instanceID1)
	assert.Nil(t, err)
	createMessageWithRoleProof := func(roleProof []byte) []byte {
		sr := signing.NewEdSigner()
		b := newMessageBuilder()
		msg := b.SetPubKey(sr.PublicKey()).SetInstanceID(instanceID1).SetRoleProof(roleProof).Sign(sr).Build()
		return mustEncode(t, msg.Message)
	}
	roleProofInbound := []byte{1, 2, 3}
	roleProofOutbound := []byte{3, 2, 1}
	serMsgInbound := createMessageWithRoleProof(roleProofInbound)
	serMsgOutbound := createMessageWithRoleProof(roleProofOutbound)
	for i := 0; i < 10; i++ {
		broker.queueMessage(context.TODO(), "not-self", serMsgInbound)
	}
	wg2.Wait()
	broker.queueMessage(context.TODO(), broker.pid, serMsgOutbound)
	wg.Done()
	tm := time.NewTimer(3 * time.Minute)
	for i := 0; i < 11; i++ {
		select {
		case x := <-outbox:
			switch i {
			case 0:
				assert.Equal(t, roleProofInbound, x.InnerMsg.RoleProof, "expected inbound msg %d", i)
			case 1:
				assert.Equal(t, roleProofOutbound, x.InnerMsg.RoleProof, "expected outbound msg %d", i)
			default:
				assert.Equal(t, roleProofInbound, x.InnerMsg.RoleProof, "expected inbound msg %d", i)
			}
		case <-tm.C:
			assert.Fail(t, "timed out waiting for message", "msg %d", i)
			return
		}
	}
	assert.Len(t, outbox, 0, "expected broker queue to be empty")
	wg3 := sync.WaitGroup{}
	wg3.Add(1)
	res := make(chan bool)
	go func() {
		timeout := time.NewTimer(time.Second)
		wg3.Done()
		select {
		case <-timeout.C:
			assert.Fail(t, "timed out waiting for channel close")
			res <- false
		case _, ok := <-broker.queueChannel:
			assert.False(t, ok, "expected channel close")
			res <- !ok
		}
	}()
	wg3.Wait()
	broker.Close()
	_, err = broker.queue.Read()
	assert.Error(t, err, "expected broker priority queue to be closed")
	assert.True(t, <-res)
}