Changeset 423 in code

Timestamp:

Oct 25, 2020, 4:47:38 PM (5 years ago)

Author:

contact

Message:

Add message store abstraction

Introduce a messageStore type, which will allow for multiple
implementations (e.g. in the DB or in-memory instead of on-disk).

The message store is per-user so that we don't need to deal with locking
and it's easier to implement per-user limits.

Location:

trunk

Files:

• downstream.go (modified) (5 diffs)
• logger.go (deleted)
• msgstore.go (added)
• upstream.go (modified) (5 diffs)
• user.go (modified) (4 diffs)

trunk/downstream.go

@@ -864 +864 @@
                         }
 
-                        lastID, err := lastMsgID(net, target, time.Now())
+                        lastID, err := dc.user.msgStore.LastMsgID(net, target, time.Now())
                         if err != nil {
                                 dc.logger.Printf("failed to get last message ID: %v", err)
@@ -877 +877 @@
 
 func (dc *downstreamConn) sendNetworkHistory(net *network) {
-        if dc.caps["draft/chathistory"] || dc.srv.LogPath == "" {
+        if dc.caps["draft/chathistory"] || dc.user.msgStore == nil {
                 return
         }
@@ -891 +891 @@
 
                 limit := 4000
-                history, err := loadHistoryLatestID(net, target, lastDelivered, limit)
+                history, err := dc.user.msgStore.LoadLatestID(net, target, lastDelivered, limit)
                 if err != nil {
                         dc.logger.Printf("failed to send implicit history for %q: %v", target, err)
@@ -1602 +1602 @@
                 }
 
-                if dc.srv.LogPath == "" {
+                if dc.user.msgStore == nil {
                         return ircError{&irc.Message{
                                 Command: irc.ERR_UNKNOWNCOMMAND,
@@ -1642 +1642 @@
                 switch subcommand {
                 case "BEFORE":
-                        history, err = loadHistoryBeforeTime(uc.network, entity, timestamp, limit)
+                        history, err = dc.user.msgStore.LoadBeforeTime(uc.network, entity, timestamp, limit)
                 case "AFTER":
-                        history, err = loadHistoryAfterTime(uc.network, entity, timestamp, limit)
+                        history, err = dc.user.msgStore.LoadAfterTime(uc.network, entity, timestamp, limit)
                 default:
                         // TODO: support LATEST, BETWEEN

trunk/upstream.go

@@ -82 +82 @@
         // set of LIST commands in progress, per downstream
         pendingLISTDownstreamSet map[uint64]struct{}
-
-        messageLoggers map[string]*messageLogger
 }
 
@@ -183 +181 @@
                 availableMemberships:     stdMemberships,
                 pendingLISTDownstreamSet: make(map[uint64]struct{}),
-                messageLoggers:           make(map[string]*messageLogger),
         }
         return uc, nil
@@ -1612 +1609 @@
 
 func (uc *upstreamConn) appendLog(entity string, msg *irc.Message) {
-        if uc.srv.LogPath == "" {
+        if uc.user.msgStore == nil {
                 return
-        }
-
-        ml, ok := uc.messageLoggers[entity]
-        if !ok {
-                ml = newMessageLogger(uc.network, entity)
-                uc.messageLoggers[entity] = ml
         }
 
@@ -1629 +1620 @@
         history, ok := uc.network.history[entity]
         if !ok {
-                lastID, err := lastMsgID(uc.network, entity, time.Now())
+                lastID, err := uc.user.msgStore.LastMsgID(uc.network, entity, time.Now())
                 if err != nil {
                         uc.logger.Printf("failed to log message: failed to get last message ID: %v", err)
@@ -1653 +1644 @@
         }
 
-        msgID, err := ml.Append(msg)
+        msgID, err := uc.user.msgStore.Append(uc.network, entity, msg)
         if err != nil {
                 uc.logger.Printf("failed to log message: %v", err)

trunk/user.go

@@ -250 +250 @@
         networks        []*network
         downstreamConns []*downstreamConn
+        msgStore        *messageStore
 
         // LIST commands in progress
@@ -262 +263 @@
 
 func newUser(srv *Server, record *User) *user {
+        var msgStore *messageStore
+        if srv.LogPath != "" {
+                msgStore = newMessageStore(srv.LogPath, record.Username)
+        }
+
         return &user{
-                User:   *record,
-                srv:    srv,
-                events: make(chan event, 64),
-                done:   make(chan struct{}),
+                User:     *record,
+                srv:      srv,
+                events:   make(chan event, 64),
+                done:     make(chan struct{}),
+                msgStore: msgStore,
         }
 }
@@ -313 +320 @@
 
 func (u *user) run() {
-        defer close(u.done)
+        defer func() {
+                if u.msgStore != nil {
+                        if err := u.msgStore.Close(); err != nil {
+                                u.srv.Logger.Printf("failed to close message store for user %q: %v", u.Username, err)
+                        }
+                }
+                close(u.done)
+        }()
 
         networks, err := u.srv.db.ListNetworks(u.ID)
@@ -460 +474 @@
         uc.network.conn = nil
 
-        for _, ml := range uc.messageLoggers {
-                if err := ml.Close(); err != nil {
-                        uc.logger.Printf("failed to close message logger: %v", err)
-                }
-        }
-
         uc.endPendingLISTs(true)