move rpc methods near the type

Author: sukunrt

comm.go

@@ -205,49 +205,3 @@ func (p *PubSub) handleSendingMessages(ctx context.Context, s network.Stream, ou
 		}
 	}
 }
-
-func rpcWithSubs(subs ...*pb.RPC_SubOpts) *RPC {
-	return &RPC{
-		RPC: pb.RPC{
-			Subscriptions: subs,
-		},
-	}
-}
-
-func rpcWithMessages(msgs ...*pb.Message) *RPC {
-	return &RPC{RPC: pb.RPC{Publish: msgs}}
-}
-
-func rpcWithMessageAndMsgID(msg *pb.Message, msgID string) *RPC {
-	return &RPC{RPC: pb.RPC{Publish: []*pb.Message{msg}}, MsgIDs: []string{msgID}}
-}
-
-func rpcWithControl(msgs []*pb.Message,
-	ihave []*pb.ControlIHave,
-	iwant []*pb.ControlIWant,
-	graft []*pb.ControlGraft,
-	prune []*pb.ControlPrune,
-	idontwant []*pb.ControlIDontWant) *RPC {
-	return &RPC{
-		RPC: pb.RPC{
-			Publish: msgs,
-			Control: &pb.ControlMessage{
-				Ihave:     ihave,
-				Iwant:     iwant,
-				Graft:     graft,
-				Prune:     prune,
-				Idontwant: idontwant,
-			},
-		},
-	}
-}
-
-func copyRPC(rpc *RPC) *RPC {
-	res := new(RPC)
-	*res = *rpc
-	if rpc.Control != nil {
-		res.Control = new(pb.ControlMessage)
-		*res.Control = *rpc.Control
-	}
-	return res
-}

pubsub.go

@@ -461,6 +461,52 @@ func (rpc *RPC) split(limit int) iter.Seq[RPC] {
 	}
 }
 
+func rpcWithSubs(subs ...*pb.RPC_SubOpts) *RPC {
+	return &RPC{
+		RPC: pb.RPC{
+			Subscriptions: subs,
+		},
+	}
+}
+
+func rpcWithMessages(msgs ...*pb.Message) *RPC {
+	return &RPC{RPC: pb.RPC{Publish: msgs}}
+}
+
+func rpcWithMessageAndMsgID(msg *pb.Message, msgID string) *RPC {
+	return &RPC{RPC: pb.RPC{Publish: []*pb.Message{msg}}, MessageIDs: []string{msgID}}
+}
+
+func rpcWithControl(msgs []*pb.Message,
+	ihave []*pb.ControlIHave,
+	iwant []*pb.ControlIWant,
+	graft []*pb.ControlGraft,
+	prune []*pb.ControlPrune,
+	idontwant []*pb.ControlIDontWant) *RPC {
+	return &RPC{
+		RPC: pb.RPC{
+			Publish: msgs,
+			Control: &pb.ControlMessage{
+				Ihave:     ihave,
+				Iwant:     iwant,
+				Graft:     graft,
+				Prune:     prune,
+				Idontwant: idontwant,
+			},
+		},
+	}
+}
+
+func copyRPC(rpc *RPC) *RPC {
+	res := new(RPC)
+	*res = *rpc
+	if rpc.Control != nil {
+		res.Control = new(pb.ControlMessage)
+		*res.Control = *rpc.Control
+	}
+	return res
+}
+
 // pbFieldNumberLT15Size is the number of bytes required to encode a protobuf
 // field number less than or equal to 15 along with its wire type. This is 1
 // byte because the protobuf encoding of field numbers is a varint encoding of:

rpc_queue_test.go

@@ -298,7 +298,7 @@ func TestRPCQueueCancellations(t *testing.T) {
 		}
 	})
 
-	t.Run("cancel duplicate", func(t *testing.T) {
+	t.Run("only one rpc cancelled", func(t *testing.T) {
 		msgs, msgIDs := getMesssages(10)
 		rpc := &RPC{
 			RPC:        pb.RPC{Publish: slices.Clone(msgs)},
@@ -326,11 +326,11 @@ func TestRPCQueueCancellations(t *testing.T) {
 		if err != nil {
 			t.Fatalf("failed to pop RPC: %v", err)
 		}
-		if !slices.Equal(msgs[3:], popped.Publish) {
-			t.Fatalf("expected popped.Publish to be %v, got %v", msgs[3:], popped.Publish)
+		if !slices.Equal(msgs, popped.Publish) {
+			t.Fatalf("expected popped.Publish to be %v, got %v", msgs, popped.Publish)
 		}
-		if !slices.Equal(msgIDs[3:], popped.MessageIDs) {
-			t.Fatalf("expected popped.MsgIDs to be %v, got %v", msgIDs[3:], popped.MessageIDs)
+		if !slices.Equal(msgIDs, popped.MessageIDs) {
+			t.Fatalf("expected popped.MsgIDs to be %v, got %v", msgIDs, popped.MessageIDs)
 		}
 		if len(q.queuedMessageIDs) != 0 {
 			t.Fatalf("expected q.queuedMsgIDs to be empty, got %v", q.queuedMessageIDs)