fix https://github.com/kataras/iris/issues/1318 with Server.SyncBroadcaster = true

Author: kataras

_examples/example/main.go

@@ -172,11 +172,12 @@ func server(upgrader neffos.Upgrader) {
 				Event:     "chat",
 				Body:      []byte(fmt.Sprintf("Client [%s] connected too.", c.ID())),
 			})
-			c.Server().Broadcast(c, neffos.Message{
-				Namespace: namespace,
-				Event:     "chat",
-				Body:      []byte(fmt.Sprintf("SECOND ONE Client [%s] connected too.", c.ID())),
-			})
+
+			// c.Server().Broadcast(c, neffos.Message{
+			// 	Namespace: namespace,
+			// 	Event:     "chat",
+			// 	Body:      []byte(fmt.Sprintf("SECOND ONE")),
+			// })
 		}
 
 		return nil

_examples/stress-test/broadcasting-1/main.go

@@ -71,7 +71,6 @@ var (
 )
 
 func main() {
-
 	// connect all and then start cron.
 	server := startServer()
 	time.Sleep(200 * time.Millisecond)
@@ -104,6 +103,7 @@ func startServer() *neffos.Server {
 
 		return username
 	}
+	server.SyncBroadcaster = true
 
 	go func() {
 		log.Fatal(http.ListenAndServe(":8080", server))

broadcaster.go

@@ -1,17 +1,17 @@
 package neffos
 
 import (
-	"context"
 	"sync"
 	"sync/atomic"
-	"time"
 	"unsafe"
 )
 
+// async broadcaster, doesn't wait for a publish to complete to all clients before any
+// next broadcast call.
 type broadcaster struct {
-	message Message
-	mu      *sync.Mutex
-	awaiter unsafe.Pointer
+	messages []Message
+	mu       *sync.Mutex
+	awaiter  unsafe.Pointer
 }
 
 func newBroadcaster() *broadcaster {
@@ -28,72 +28,26 @@ func (b *broadcaster) getAwaiter() <-chan struct{} {
 	return *((*chan struct{})(ptr))
 }
 
-func (b *broadcaster) broadcast(msg Message) {
+func (b *broadcaster) broadcast(msgs []Message) {
 	b.mu.Lock()
-	b.message = msg
+	b.messages = msgs
 	b.mu.Unlock()
 
 	ch := make(chan struct{})
 	old := atomic.SwapPointer(&b.awaiter, unsafe.Pointer(&ch))
 	close(*(*chan struct{})(old))
 }
 
-// lock required.
-func (b *broadcaster) wait() (msg Message) {
+func (b *broadcaster) waitUntilClosed(closeCh <-chan struct{}) (msgs []Message, ok bool) {
 	ch := b.getAwaiter()
 	b.mu.Unlock()
-	<-ch
-	msg = b.message
-	b.mu.Lock()
-
-	return
-}
+	select {
+	case <-ch:
+		msgs = b.messages[:]
+		ok = true
+	case <-closeCh:
+	}
 
-func (b *broadcaster) subscribe(fn func(<-chan struct{})) {
-	ch := b.getAwaiter()
-	b.mu.Unlock()
-	fn(ch)
 	b.mu.Lock()
-}
-
-func (b *broadcaster) waitUntilClosed(closeCh <-chan struct{}) (msg Message, ok bool) {
-	b.subscribe(func(ch <-chan struct{}) {
-		select {
-		case <-ch:
-			msg = b.message
-			ok = true
-		case <-closeCh:
-		}
-	})
-
-	return
-}
-
-func (b *broadcaster) waitUntil(timeout time.Duration) (msg Message, ok bool) {
-	timer := time.NewTimer(timeout)
-	defer timer.Stop()
-
-	b.subscribe(func(ch <-chan struct{}) {
-		select {
-		case <-ch:
-			msg = b.message
-			ok = true
-		case <-timer.C:
-		}
-	})
-
-	return
-}
-
-func (b *broadcaster) waitWithContext(ctx context.Context) (msg Message, err error) {
-	b.subscribe(func(ch <-chan struct{}) {
-		select {
-		case <-ch:
-			msg = b.message
-		case <-ctx.Done():
-			err = ctx.Err()
-		}
-	})
-
 	return
 }

conn.go

@@ -88,7 +88,7 @@ type Conn struct {
 
 	// used to fire `conn#Close` once.
 	closed *uint32
-	// useful to terminate the broadcaster, see `Server#ServeHTTP.waitMessage`.
+	// useful to terminate the broadcaster, see `Server#ServeHTTP.waitMessages`.
 	closeCh chan struct{}
 }
 

server.go

@@ -49,6 +49,17 @@ type Server struct {
 	IDGenerator   IDGenerator
 	StackExchange StackExchange
 
+	// If `StackExchange` is set then this field is ignored.
+	//
+	// It overrides the default behavior(when no StackExchange is not used)
+	// which publishes a message independently.
+	// In short the default behavior doesn't wait for a message to be published to all clients
+	// before any next broadcast call.
+	//
+	// Therefore, if set to true,
+	// each broadcast call will publish its own message(s) by order.
+	SyncBroadcaster bool
+
 	mu         sync.RWMutex
 	namespaces Namespaces
 
@@ -58,14 +69,16 @@ type Server struct {
 
 	count uint64
 
-	connections map[*Conn]struct{}
-	connect     chan *Conn
-	disconnect  chan *Conn
-	actions     chan action
+	connections       map[*Conn]struct{}
+	connect           chan *Conn
+	disconnect        chan *Conn
+	actions           chan action
+	broadcastMessages chan []Message
+
 	broadcaster *broadcaster
+
 	// messages that this server must waits
-	// for a reply from one of its own connections(see `waitMessage`)
-	// or TODO: from cloud (see `StackExchange.PublishAndWait`).
+	// for a reply from one of its own connections(see `waitMessages`).
 	waitingMessages      map[string]chan Message
 	waitingMessagesMutex sync.RWMutex
 
@@ -94,22 +107,21 @@ func New(upgrader Upgrader, connHandler ConnHandler) *Server {
 	readTimeout, writeTimeout := getTimeouts(connHandler)
 	namespaces := connHandler.GetNamespaces()
 	s := &Server{
-		uuid:            uuid.Must(uuid.NewV4()).String(),
-		upgrader:        upgrader,
-		namespaces:      namespaces,
-		readTimeout:     readTimeout,
-		writeTimeout:    writeTimeout,
-		connections:     make(map[*Conn]struct{}),
-		connect:         make(chan *Conn, 1),
-		disconnect:      make(chan *Conn),
-		actions:         make(chan action),
-		broadcaster:     newBroadcaster(),
-		waitingMessages: make(map[string]chan Message),
-		IDGenerator:     DefaultIDGenerator,
+		uuid:              uuid.Must(uuid.NewV4()).String(),
+		upgrader:          upgrader,
+		namespaces:        namespaces,
+		readTimeout:       readTimeout,
+		writeTimeout:      writeTimeout,
+		connections:       make(map[*Conn]struct{}),
+		connect:           make(chan *Conn, 1),
+		disconnect:        make(chan *Conn),
+		actions:           make(chan action),
+		broadcastMessages: make(chan []Message),
+		broadcaster:       newBroadcaster(),
+		waitingMessages:   make(map[string]chan Message),
+		IDGenerator:       DefaultIDGenerator,
 	}
 
-	//	s.broadcastCond = sync.NewCond(&s.broadcastMu)
-
 	go s.start()
 
 	return s
@@ -171,6 +183,10 @@ func (s *Server) start() {
 					s.StackExchange.OnDisconnect(c)
 				}
 			}
+		case msgs := <-s.broadcastMessages:
+			for c := range s.connections {
+				publishMessages(c, msgs)
+			}
 		case act := <-s.actions:
 			for c := range s.connections {
 				act.call(c)
@@ -312,9 +328,9 @@ func (s *Server) Upgrade(
 		c.ReconnectTries, _ = strconv.Atoi(retriesHeaderValue)
 	}
 
-	if !s.usesStackExchange() {
+	if !s.usesStackExchange() && !s.SyncBroadcaster {
 		go func(c *Conn) {
-			for s.waitMessage(c) {
+			for s.waitMessages(c) {
 			}
 		}(c)
 	}
@@ -380,34 +396,6 @@ func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	s.Upgrade(w, r, nil, nil)
 }
 
-func (s *Server) waitMessage(c *Conn) bool {
-	s.broadcaster.mu.Lock()
-	defer s.broadcaster.mu.Unlock()
-
-	msg, ok := s.broadcaster.waitUntilClosed(c.closeCh)
-	if !ok {
-		return false
-	}
-
-	if msg.from == c.ID() {
-		// if the message is not supposed to return back to any connection with this ID.
-		return true
-	}
-
-	// if "To" field is given then send to a specific connection.
-	if msg.To != "" && msg.To != c.ID() {
-		return true
-	}
-
-	// c.Write may fail if the message is not supposed to end to this client
-	// but the connection should be still open in order to continue.
-	if !c.Write(msg) && c.IsClosed() {
-		return false
-	}
-
-	return true
-}
-
 // GetTotalConnections returns the total amount of the connected connections to the server, it's fast
 // and can be used as frequently as needed.
 func (s *Server) GetTotalConnections() uint64 {
@@ -438,6 +426,41 @@ func (s *Server) Do(fn func(*Conn), async bool) {
 	}
 }
 
+func publishMessages(c *Conn, msgs []Message) bool {
+	for _, msg := range msgs {
+		if msg.from == c.ID() {
+			// if the message is not supposed to return back to any connection with this ID.
+
+			return true
+		}
+
+		// if "To" field is given then send to a specific connection.
+		if msg.To != "" && msg.To != c.ID() {
+			return true
+		}
+
+		// c.Write may fail if the message is not supposed to end to this client
+		// but the connection should be still open in order to continue.
+		if !c.Write(msg) && c.IsClosed() {
+			return false
+		}
+	}
+
+	return true
+}
+
+func (s *Server) waitMessages(c *Conn) bool {
+	s.broadcaster.mu.Lock()
+	defer s.broadcaster.mu.Unlock()
+
+	msgs, ok := s.broadcaster.waitUntilClosed(c.closeCh)
+	if !ok {
+		return false
+	}
+
+	return publishMessages(c, msgs)
+}
+
 type stringerValue struct{ v string }
 
 func (s stringerValue) String() string { return s.v }
@@ -452,7 +475,7 @@ func (s stringerValue) String() string { return s.v }
 //  neffos.Message{Namespace: "default", Room: "roomName or empty", Event: "chat", Body: [...]})
 func Exclude(connID string) fmt.Stringer { return stringerValue{connID} }
 
-// Broadcast method is fast and does not block any new incoming connection,
+// Broadcast method is fast and does not block any new incoming connection by-default,
 // it can be used as frequently as needed. Use the "msg"'s Namespace, or/and Event or/and Room to broadcast
 // to a specific type of connection collectives.
 //
@@ -465,34 +488,45 @@ func Exclude(connID string) fmt.Stringer { return stringerValue{connID} }
 // nsConn.Conn.Server().Broadcast(
 //	nsConn OR nil,
 //  neffos.Message{Namespace: "default", Room: "roomName or empty", Event: "chat", Body: [...]})
-func (s *Server) Broadcast(exceptSender fmt.Stringer, msg Message) {
+//
+// Note that it if `StackExchange` is nil then its default behavior
+// doesn't wait for a publish to complete to all clients before any
+// next broadcast call. To change that behavior set the `Server.SyncBroadcaster` to true
+// before server start.
+func (s *Server) Broadcast(exceptSender fmt.Stringer, msgs ...Message) {
+
 	if exceptSender != nil {
+		var fromExplicit, from string
+
 		switch c := exceptSender.(type) {
 		case *Conn:
-			msg.FromExplicit = c.serverConnID
+			fromExplicit = c.serverConnID
 		case *NSConn:
-			msg.FromExplicit = c.Conn.serverConnID
+			fromExplicit = c.Conn.serverConnID
 		default:
-			msg.from = exceptSender.String()
+			from = exceptSender.String()
 		}
 
-		// msg.from = exceptSender.String()
+		for i := range msgs {
+			if from != "" {
+				msgs[i].from = from
+			} else {
+				msgs[i].FromExplicit = fromExplicit
+			}
+		}
 	}
 
-	// s.broadcast <- msg
-
-	// s.broadcastMu.Lock()
-	// s.broadcastMessage = msg
-	// s.broadcastMu.Unlock()
-
-	// s.broadcastCond.Broadcast()
-
 	if s.usesStackExchange() {
-		s.StackExchange.Publish(msg)
+		s.StackExchange.Publish(msgs)
+		return
+	}
+
+	if s.SyncBroadcaster {
+		s.broadcastMessages <- msgs
 		return
 	}
 
-	s.broadcaster.broadcast(msg)
+	s.broadcaster.broadcast(msgs)
 }
 
 // Ask is like `Broadcast` but it blocks until a response