Relay WebSocket library refactor

arbnode/inbox_tracker.go

@@ -22,6 +22,7 @@ import (
 	"github.com/offchainlabs/nitro/arbstate"
 	"github.com/offchainlabs/nitro/arbutil"
 	"github.com/offchainlabs/nitro/broadcaster"
+	m "github.com/offchainlabs/nitro/broadcaster/message"
 	"github.com/offchainlabs/nitro/staker"
 	"github.com/offchainlabs/nitro/util/containers"
 )
@@ -240,7 +241,7 @@ func (t *InboxTracker) PopulateFeedBacklog(broadcastServer *broadcaster.Broadcas
 	if err != nil {
 		return fmt.Errorf("error getting tx streamer message count: %w", err)
 	}
-	var feedMessages []*broadcaster.BroadcastFeedMessage
+	var feedMessages []*m.BroadcastFeedMessage
 	for seqNum := startMessage; seqNum < messageCount; seqNum++ {
 		message, err := t.txStreamer.GetMessage(seqNum)
 		if err != nil {

arbnode/transaction_streamer.go

@@ -32,6 +32,7 @@ import (
 	"github.com/offchainlabs/nitro/arbos/arbostypes"
 	"github.com/offchainlabs/nitro/arbutil"
 	"github.com/offchainlabs/nitro/broadcaster"
+	m "github.com/offchainlabs/nitro/broadcaster/message"
 	"github.com/offchainlabs/nitro/execution"
 	"github.com/offchainlabs/nitro/staker"
 	"github.com/offchainlabs/nitro/util/arbmath"
@@ -426,7 +427,7 @@ func (s *TransactionStreamer) AddMessages(pos arbutil.MessageIndex, messagesAreC
 	return s.AddMessagesAndEndBatch(pos, messagesAreConfirmed, messages, nil)
 }
 
-func (s *TransactionStreamer) AddBroadcastMessages(feedMessages []*broadcaster.BroadcastFeedMessage) error {
+func (s *TransactionStreamer) AddBroadcastMessages(feedMessages []*m.BroadcastFeedMessage) error {
 	if len(feedMessages) == 0 {
 		return nil
 	}

broadcastclient/broadcastclient.go

@@ -27,7 +27,7 @@ import (
 	"github.com/ethereum/go-ethereum/metrics"
 
 	"github.com/offchainlabs/nitro/arbutil"
-	"github.com/offchainlabs/nitro/broadcaster"
+	m "github.com/offchainlabs/nitro/broadcaster/message"
 	"github.com/offchainlabs/nitro/util/contracts"
 	"github.com/offchainlabs/nitro/util/signature"
 	"github.com/offchainlabs/nitro/util/stopwaiter"
@@ -117,7 +117,7 @@ var DefaultTestConfig = Config{
 }
 
 type TransactionStreamerInterface interface {
-	AddBroadcastMessages(feedMessages []*broadcaster.BroadcastFeedMessage) error
+	AddBroadcastMessages(feedMessages []*m.BroadcastFeedMessage) error
 }
 
 type BroadcastClient struct {
@@ -381,7 +381,7 @@ func (bc *BroadcastClient) startBackgroundReader(earlyFrameData io.Reader) {
 			backoffDuration = bc.config().ReconnectInitialBackoff
 
 			if msg != nil {
-				res := broadcaster.BroadcastMessage{}
+				res := m.BroadcastMessage{}
 				err = json.Unmarshal(msg, &res)
 				if err != nil {
 					log.Error("error unmarshalling message", "msg", msg, "err", err)
@@ -483,7 +483,7 @@ func (bc *BroadcastClient) StopAndWait() {
 	}
 }
 
-func (bc *BroadcastClient) isValidSignature(ctx context.Context, message *broadcaster.BroadcastFeedMessage) error {
+func (bc *BroadcastClient) isValidSignature(ctx context.Context, message *m.BroadcastFeedMessage) error {
 	if bc.config().Verify.Dangerous.AcceptMissing && bc.sigVerifier == nil {
 		// Verifier disabled
 		return nil

broadcastclient/broadcastclient_test.go

@@ -23,6 +23,7 @@ import (
 	"github.com/offchainlabs/nitro/arbos/arbostypes"
 	"github.com/offchainlabs/nitro/arbutil"
 	"github.com/offchainlabs/nitro/broadcaster"
+	m "github.com/offchainlabs/nitro/broadcaster/message"
 	"github.com/offchainlabs/nitro/util/contracts"
 	"github.com/offchainlabs/nitro/util/signature"
 	"github.com/offchainlabs/nitro/util/testhelpers"
@@ -178,20 +179,20 @@ func TestInvalidSignature(t *testing.T) {
 }
 
 type dummyTransactionStreamer struct {
-	messageReceiver chan broadcaster.BroadcastFeedMessage
+	messageReceiver chan m.BroadcastFeedMessage
 	chainId         uint64
 	sequencerAddr   *common.Address
 }
 
 func NewDummyTransactionStreamer(chainId uint64, sequencerAddr *common.Address) *dummyTransactionStreamer {
 	return &dummyTransactionStreamer{
-		messageReceiver: make(chan broadcaster.BroadcastFeedMessage),
+		messageReceiver: make(chan m.BroadcastFeedMessage),
 		chainId:         chainId,
 		sequencerAddr:   sequencerAddr,
 	}
 }
 
-func (ts *dummyTransactionStreamer) AddBroadcastMessages(feedMessages []*broadcaster.BroadcastFeedMessage) error {
+func (ts *dummyTransactionStreamer) AddBroadcastMessages(feedMessages []*m.BroadcastFeedMessage) error {
 	for _, feedMessage := range feedMessages {
 		ts.messageReceiver <- *feedMessage
 	}

broadcastclients/broadcastclients.go

@@ -12,7 +12,7 @@ import (
 
 	"github.com/offchainlabs/nitro/arbutil"
 	"github.com/offchainlabs/nitro/broadcastclient"
-	"github.com/offchainlabs/nitro/broadcaster"
+	m "github.com/offchainlabs/nitro/broadcaster/message"
 	"github.com/offchainlabs/nitro/util/contracts"
 	"github.com/offchainlabs/nitro/util/stopwaiter"
 )
@@ -25,14 +25,14 @@ const PRIMARY_FEED_UPTIME = time.Minute * 10
 
 type Router struct {
 	stopwaiter.StopWaiter
-	messageChan                 chan broadcaster.BroadcastFeedMessage
+	messageChan                 chan m.BroadcastFeedMessage
 	confirmedSequenceNumberChan chan arbutil.MessageIndex
 
 	forwardTxStreamer       broadcastclient.TransactionStreamerInterface
 	forwardConfirmationChan chan arbutil.MessageIndex
 }
 
-func (r *Router) AddBroadcastMessages(feedMessages []*broadcaster.BroadcastFeedMessage) error {
+func (r *Router) AddBroadcastMessages(feedMessages []*m.BroadcastFeedMessage) error {
 	for _, feedMessage := range feedMessages {
 		r.messageChan <- *feedMessage
 	}
@@ -67,7 +67,7 @@ func NewBroadcastClients(
 	}
 	newStandardRouter := func() *Router {
 		return &Router{
-			messageChan:                 make(chan broadcaster.BroadcastFeedMessage, ROUTER_QUEUE_SIZE),
+			messageChan:                 make(chan m.BroadcastFeedMessage, ROUTER_QUEUE_SIZE),
 			confirmedSequenceNumberChan: make(chan arbutil.MessageIndex, ROUTER_QUEUE_SIZE),
 			forwardTxStreamer:           txStreamer,
 			forwardConfirmationChan:     confirmedSequenceNumberListener,
@@ -152,15 +152,15 @@ func (bcs *BroadcastClients) Start(ctx context.Context) {
 		defer stopSecondaryFeedTimer.Stop()
 		defer primaryFeedIsDownTimer.Stop()
 
-		msgHandler := func(msg broadcaster.BroadcastFeedMessage, router *Router) error {
+		msgHandler := func(msg m.BroadcastFeedMessage, router *Router) error {
 			if _, ok := recentFeedItemsNew[msg.SequenceNumber]; ok {
 				return nil
 			}
 			if _, ok := recentFeedItemsOld[msg.SequenceNumber]; ok {
 				return nil
 			}
 			recentFeedItemsNew[msg.SequenceNumber] = time.Now()
-			if err := router.forwardTxStreamer.AddBroadcastMessages([]*broadcaster.BroadcastFeedMessage{&msg}); err != nil {
+			if err := router.forwardTxStreamer.AddBroadcastMessages([]*m.BroadcastFeedMessage{&msg}); err != nil {
 				return err
 			}
 			return nil