test(broadcast): added tests for broadcast

Author: aleksander-vedvik

broadcast.go

@@ -2,9 +2,6 @@ package gorums
 
 import (
 	"context"
-	"errors"
-	"log"
-	"sync"
 	"time"
 
 	"github.com/google/uuid"
@@ -14,24 +11,20 @@ import (
 )
 
 type broadcastServer struct {
-	sync.RWMutex
 	id              string
 	addr            string
 	broadcastedMsgs map[string]map[string]bool
 	handlers        map[string]broadcastFunc
 	broadcastChan   chan *broadcastMsg
-	responseChan    chan responseMsg
+	responseChan    chan *responseMsg
 	clientHandlers  map[string]func(addr, broadcastID string, req protoreflect.ProtoMessage, opts ...grpc.CallOption) (any, error)
 	bNew            iBroadcastStruct
 	timeout         time.Duration
 	clientReqs      *RequestMap
 	view            serverView
 	middlewares     []func(BroadcastMetadata) error
 	stopChan        chan struct{}
-	//mutex          sync.RWMutex
-	//returnedToClientMsgs map[string]bool
-	//clientReqs      map[string]*clientRequest
-	//clientReqsMutex sync.Mutex
+	async           bool
 }
 
 func newBroadcastServer() *broadcastServer {
@@ -41,12 +34,10 @@ func newBroadcastServer() *broadcastServer {
 		clientHandlers:  make(map[string]func(addr, broadcastID string, req protoreflect.ProtoMessage, opts ...grpc.CallOption) (any, error)),
 		broadcastChan:   make(chan *broadcastMsg, 1000),
 		handlers:        make(map[string]broadcastFunc),
-		responseChan:    make(chan responseMsg),
+		responseChan:    make(chan *responseMsg),
 		clientReqs:      NewRequestMap(),
 		middlewares:     make([]func(BroadcastMetadata) error, 0),
 		stopChan:        make(chan struct{}, 0),
-		//returnedToClientMsgs: make(map[string]bool),
-		//clientReqs:    make(map[string]*clientRequest),
 	}
 }
 
@@ -77,26 +68,25 @@ func (srv *broadcastServer) handleClientResponses() {
 	}
 }
 
-func (srv *broadcastServer) handle(response responseMsg) {
+func (srv *broadcastServer) handle(response *responseMsg) {
 	broadcastID := response.getBroadcastID()
-	req, handled := srv.clientReqs.GetSet(broadcastID)
+	req, handled := srv.clientReqs.GetAndSetHandled(broadcastID)
 	if handled {
 		// this server has not received a request directly from a client
 		// hence, the response should be ignored
 		// already handled and can not be removed yet. It is possible to get duplicates.
 		return
 	}
-	select {
-	case <-req.ctx.Done():
-		// client request has been cancelled by client
-		log.Println("CLIENT REQUEST HAS BEEN CANCELLED")
-		return
-	default:
-	}
+	//select {
+	//case <-req.ctx.Done():
+	//	// client request has been cancelled by client
+	//	log.Println("CLIENT REQUEST HAS BEEN CANCELLED")
+	//	return
+	//default:
+	//}
 	if !response.valid() {
 		// the response is old and should have timed out, but may not due to scheduling.
 		// the timeout msg should arrive soon.
-		log.Println("NOT VALID")
 		return
 	}
 	if handler, ok := srv.clientHandlers[req.metadata.BroadcastMsg.OriginMethod]; ok && req.metadata.BroadcastMsg.OriginAddr != "" {
@@ -121,17 +111,17 @@ func (srv *broadcastServer) clientReturn(resp ResponseTypes, err error, metadata
 }
 
 func (srv *broadcastServer) returnToClient(broadcastID string, resp ResponseTypes, err error) {
-	srv.Lock()
-	defer srv.Unlock()
+	//srv.Lock()
+	//defer srv.Unlock()
 	if !srv.alreadyReturnedToClient(broadcastID) {
 		srv.responseChan <- newResponseMessage(resp, err, broadcastID, clientResponse, srv.timeout)
 	}
 }
 
-func (srv *broadcastServer) timeoutClientResponse(ctx ServerCtx, in *Message, finished chan<- *Message) {
-	time.Sleep(srv.timeout)
-	srv.responseChan <- newResponseMessage(protoreflect.ProtoMessage(nil), errors.New("server timed out"), in.Metadata.BroadcastMsg.GetBroadcastID(), timeout, srv.timeout)
-}
+//func (srv *broadcastServer) timeoutClientResponse(ctx ServerCtx, in *Message, finished chan<- *Message) {
+//	time.Sleep(srv.timeout)
+//	srv.responseChan <- newResponseMessage(protoreflect.ProtoMessage(nil), errors.New("server timed out"), in.Metadata.BroadcastMsg.GetBroadcastID(), timeout, srv.timeout)
+//}
 
 func (srv *broadcastServer) alreadyReturnedToClient(broadcastID string) bool {
 	req, ok := srv.clientReqs.Get(broadcastID)

broadcastTypes.go

@@ -38,14 +38,6 @@ type defaultImplementationFunc[T RequestTypes, V ResponseTypes] func(ServerCtx,
 
 type implementationFunc[T RequestTypes, V iBroadcastStruct] func(ServerCtx, T, V)
 
-type responseMsg interface {
-	getResponse() ResponseTypes
-	getError() error
-	getBroadcastID() string
-	valid() bool
-	getType() respType
-}
-
 type respType int
 
 const (
@@ -55,7 +47,7 @@ const (
 	done
 )
 
-type responseMessage struct {
+type responseMsg struct {
 	response    ResponseTypes
 	err         error
 	broadcastID string
@@ -64,8 +56,8 @@ type responseMessage struct {
 	respType    respType
 }
 
-func newResponseMessage(response ResponseTypes, err error, broadcastID string, respType respType, ttl time.Duration) *responseMessage {
-	return &responseMessage{
+func newResponseMessage(response ResponseTypes, err error, broadcastID string, respType respType, ttl time.Duration) *responseMsg {
+	return &responseMsg{
 		response:    response,
 		err:         err,
 		broadcastID: broadcastID,
@@ -75,23 +67,23 @@ func newResponseMessage(response ResponseTypes, err error, broadcastID string, r
 	}
 }
 
-func (r *responseMessage) getResponse() ResponseTypes {
+func (r *responseMsg) getResponse() ResponseTypes {
 	return r.response
 }
 
-func (r *responseMessage) getError() error {
+func (r *responseMsg) getError() error {
 	return r.err
 }
 
-func (r *responseMessage) getBroadcastID() string {
+func (r *responseMsg) getBroadcastID() string {
 	return r.broadcastID
 }
 
-func (r *responseMessage) valid() bool {
+func (r *responseMsg) valid() bool {
 	return r.respType == clientResponse && time.Since(r.timestamp) <= r.ttl
 }
 
-func (r *responseMessage) getType() respType {
+func (r *responseMsg) getType() respType {
 	return r.respType
 }
 
@@ -260,13 +252,13 @@ func (list *RequestMap) Get(identifier string) (clientRequest, bool) {
 	return elem, ok
 }
 
-func (list *RequestMap) Set(identifier string, elem clientRequest) {
-	list.mutex.Lock()
-	defer list.mutex.Unlock()
-	list.data[identifier] = elem
-}
+//func (list *RequestMap) Set(identifier string, elem clientRequest) {
+//	list.mutex.Lock()
+//	defer list.mutex.Unlock()
+//	list.data[identifier] = elem
+//}
 
-func (list *RequestMap) GetSet(identifier string) (clientRequest, bool) {
+func (list *RequestMap) GetAndSetHandled(identifier string) (clientRequest, bool) {
 	list.mutex.Lock()
 	defer list.mutex.Unlock()
 	elem, ok := list.data[identifier]

broadcast_test.go

@@ -62,7 +62,19 @@ func (b *testBroadcast) Broadcast(req *testBroadcastRequest, opts ...BroadcastOp
 	for _, opt := range opts {
 		opt(&data)
 	}
-	b.sp.BroadcastHandler("broadcast", req, b.metadata, data)
+	b.sp.BroadcastHandler("Broadcast", req, b.metadata, data)
+}
+
+func (b *testBroadcast) CanBroadcast(req *testBroadcastRequest, opts ...BroadcastOption) {
+	data := NewBroadcastOptions()
+	for _, opt := range opts {
+		opt(&data)
+	}
+	b.sp.BroadcastHandler("CanBroadcast", req, b.metadata, data)
+}
+
+func (b *testBroadcast) SendToClient(resp protoreflect.ProtoMessage, err error) {
+	b.sp.ReturnToClientHandler(resp, err, b.metadata)
 }
 
 //
@@ -175,6 +187,10 @@ type testBroadcastServer struct {
 	req     *testBroadcastRequest
 }
 
+func (srv *testBroadcastServer) SendToClient(resp protoreflect.ProtoMessage, err error, broadcastID string) {
+	srv.RetToClient(resp, err, broadcastID)
+}
+
 func newTestBroadcastServer() *testBroadcastServer {
 	srv := &testBroadcastServer{
 		Server: NewServer(),
@@ -194,6 +210,18 @@ func (srv *testBroadcastServer) Broadcast(ctx ServerCtx, request *testBroadcastR
 	//broadcast.Broadcast(&testBroadcastRequest{})
 }
 
+func (srv *testBroadcastServer) CanBroadcast(ctx ServerCtx, request *testBroadcastRequest, broadcast *testBroadcast) {
+	srv.numMsgs++
+	srv.req = request
+	go broadcast.Broadcast(&testBroadcastRequest{})
+}
+
+func (srv *testBroadcastServer) SendToClientBroadcast(ctx ServerCtx, request *testBroadcastRequest, broadcast *testBroadcast) {
+	srv.numMsgs++
+	srv.req = request
+	go broadcast.SendToClient(&testBroadcastRequest{}, nil)
+}
+
 func createReq(val string) (ServerCtx, *Message, chan<- *Message) {
 	var mut sync.Mutex
 	mut.Lock()
@@ -216,16 +244,16 @@ func createReq(val string) (ServerCtx, *Message, chan<- *Message) {
 }
 
 func TestBroadcastHandler(t *testing.T) {
-	handlerName := "broadcast"
+	handlerName := "Broadcast"
 
 	// create a server
 	srv := newTestBroadcastServer()
 	// register the broadcast handler. Similar to proto option: broadcast
 	srv.RegisterHandler(handlerName, BroadcastHandler(srv.Broadcast, srv.Server))
 
-	// create a request
 	vals := []string{"test1", "test2", "test3"}
 	for _, val := range vals {
+		// create the request
 		srvCtx, req, finished := createReq(val)
 		// call the server handler
 		srv.srv.handlers[handlerName](srvCtx, req, finished)
@@ -235,3 +263,55 @@ func TestBroadcastHandler(t *testing.T) {
 		}
 	}
 }
+
+func TestCanBroadcastHandler(t *testing.T) {
+	handlerName := "CanBroadcast"
+
+	// create a server
+	srv := newTestBroadcastServer()
+	// register the broadcast handler. Similar to proto option: broadcast
+	srv.RegisterHandler(handlerName, BroadcastHandler(srv.CanBroadcast, srv.Server))
+
+	vals := []string{"test1", "test2", "test3"}
+	for _, val := range vals {
+		// create the request
+		srvCtx, req, finished := createReq(val)
+		// call the server handler
+		srv.srv.handlers[handlerName](srvCtx, req, finished)
+		broadcastMsg := <-srv.broadcastSrv.broadcastChan
+
+		if broadcastMsg == nil {
+			t.Errorf("broadcastMsg should not be nil")
+			continue
+		}
+		if broadcastMsg.broadcastID != req.Metadata.BroadcastMsg.BroadcastID {
+			t.Errorf("broadcastID = %v, expected %s", broadcastMsg.broadcastID, req.Metadata.BroadcastMsg.BroadcastID)
+		}
+	}
+}
+
+func TestBroadcastSendToClient(t *testing.T) {
+	handlerName := "SendToClientBroadcast"
+
+	// create a server
+	srv := newTestBroadcastServer()
+	// register the broadcast handler. Similar to proto option: broadcast
+	srv.RegisterHandler(handlerName, BroadcastHandler(srv.SendToClientBroadcast, srv.Server))
+
+	vals := []string{"test1", "test2", "test3"}
+	for _, val := range vals {
+		// create the request
+		srvCtx, req, finished := createReq(val)
+		// call the server handler
+		srv.srv.handlers[handlerName](srvCtx, req, finished)
+		responseMsg := <-srv.broadcastSrv.responseChan
+
+		if responseMsg == nil {
+			t.Errorf("responseMsg should not be nil")
+			continue
+		}
+		if responseMsg.getBroadcastID() != req.Metadata.BroadcastMsg.BroadcastID {
+			t.Errorf("broadcastID = %v, expected %s", responseMsg.getBroadcastID(), req.Metadata.BroadcastMsg.BroadcastID)
+		}
+	}
+}

cmd/protoc-gen-gorums/dev/zorums_broadcast_gorums.pb.go

@@ -7,11 +7,11 @@ var broadcastVar = `
 var broadcastSignature = `func (b *Broadcast) {{.Method.GoName}}(req *{{in .GenFile .Method}}, opts... gorums.BroadcastOption) {`
 
 var broadcastBody = `
-	data := gorums.NewBroadcastOptions()
+	options := gorums.NewBroadcastOptions()
 	for _, opt := range opts {
-		opt(&data)
+		opt(&options)
 	}
-	b.sp.BroadcastHandler("{{.Method.Desc.FullName}}", req, b.metadata, data)
+	b.sp.BroadcastHandler("{{.Method.Desc.FullName}}", req, b.metadata, options)
 }
 `