Changeset 103 in code for trunk/user.go

Timestamp:

Mar 16, 2020, 11:44:59 AM (5 years ago)

Author:

contact

Message:

Per-user dispatcher goroutine

This allows message handlers to read upstream/downstream connection
information without causing any race condition.

References: ​https://todo.sr.ht/~emersion/soju/1

File:

• trunk/user.go (modified) (5 diffs)

trunk/user.go

@@ -4 +4 @@
         "sync"
         "time"
+
+        "gopkg.in/irc.v3"
 )
+
+type upstreamIncomingMessage struct {
+        msg *irc.Message
+        uc  *upstreamConn
+}
+
+type downstreamIncomingMessage struct {
+        msg *irc.Message
+        dc  *downstreamConn
+}
 
 type network struct {
@@ -41 +53 @@
                 net.user.lock.Unlock()
 
-                if err := uc.readMessages(); err != nil {
+                if err := uc.readMessages(net.user.upstreamIncoming); err != nil {
                         uc.logger.Printf("failed to handle messages: %v", err)
                 }
@@ -56 +68 @@
         srv *Server
 
+        upstreamIncoming   chan upstreamIncomingMessage
+        downstreamIncoming chan downstreamIncomingMessage
+
         lock            sync.Mutex
         networks        []*network
@@ -63 +78 @@
 func newUser(srv *Server, record *User) *user {
         return &user{
-                User: *record,
-                srv:  srv,
+                User:               *record,
+                srv:                srv,
+                upstreamIncoming:   make(chan upstreamIncomingMessage, 64),
+                downstreamIncoming: make(chan downstreamIncomingMessage, 64),
         }
 }
@@ -120 +137 @@
         }
         u.lock.Unlock()
+
+        for {
+                select {
+                case upstreamMsg := <-u.upstreamIncoming:
+                        msg, uc := upstreamMsg.msg, upstreamMsg.uc
+                        if err := uc.handleMessage(msg); err != nil {
+                                uc.logger.Printf("failed to handle message %q: %v", msg, err)
+                        }
+                case downstreamMsg := <-u.downstreamIncoming:
+                        msg, dc := downstreamMsg.msg, downstreamMsg.dc
+                        err := dc.handleMessage(msg)
+                        if ircErr, ok := err.(ircError); ok {
+                                ircErr.Message.Prefix = dc.srv.prefix()
+                                dc.SendMessage(ircErr.Message)
+                        } else if err != nil {
+                                dc.logger.Printf("failed to handle message %q: %v", msg, err)
+                                dc.Close()
+                        }
+                }
+        }
 }