Changeset 51 in code for trunk

Timestamp:

Feb 7, 2020, 4:35:57 PM (5 years ago)

Author:

contact

Message:

Add names to consumers

Location:

trunk

Files:

• downstream.go (modified) (2 diffs)
• ring.go (modified) (3 diffs)

trunk/downstream.go

@@ -129 +129 @@
                 }
                 u.lock.Unlock()
+
+                // TODO: figure out a better way to advance the ring buffer consumer cursor
+                u.forEachUpstream(func(uc *upstreamConn) {
+                        // TODO: let clients specify the ring buffer name in their username
+                        uc.ring.Consumer("").Reset()
+                })
         }
 
@@ -233 +239 @@
                 }
 
-                consumer := uc.ring.Consumer()
+                // TODO: let clients specify the ring buffer name in their username
+                consumer := uc.ring.Consumer("")
                 for {
+                        // TODO: these messages will get lost if the connection is closed
                         msg := consumer.Consume()
                         if msg == nil {

trunk/ring.go

@@ -11 +11 @@
         cap, cur uint64
 
-        consumers []RingConsumer
+        consumers map[string]*RingConsumer
 }
 
 func NewRing(capacity int) *Ring {
         return &Ring{
-                buffer: make([]*irc.Message, capacity),
-                cap:    uint64(capacity),
+                buffer:    make([]*irc.Message, capacity),
+                cap:       uint64(capacity),
+                consumers: make(map[string]*RingConsumer),
         }
 }
@@ -27 +28 @@
 }
 
-func (r *Ring) Consumer() *RingConsumer {
-        return &RingConsumer{
+func (r *Ring) Consumer(name string) *RingConsumer {
+        consumer, ok := r.consumers[name]
+        if ok {
+                return consumer
+        }
+
+        consumer = &RingConsumer{
                 ring: r,
-                cur:  0, // r.cur
+                cur:  r.cur,
         }
+        r.consumers[name] = consumer
+        return consumer
 }
 
@@ -70 +78 @@
         return msg
 }
+
+func (rc *RingConsumer) Reset() {
+        rc.cur = rc.ring.cur
+}