Changeset 143 in code for trunk/downstream.go

Timestamp:

Mar 25, 2020, 9:53:08 AM (5 years ago)

Author:

contact

Message:

Move upstreamConn.ring to network

This handles upstream disconnection and re-connection better.

File:

• trunk/downstream.go (modified) (6 diffs)

trunk/downstream.go

@@ -700 +700 @@
                         }
                 }
-
+        })
+
+        dc.forEachNetwork(func(net *network) {
                 historyName := dc.rawUsername
 
@@ -706 +708 @@
                 var seqPtr *uint64
                 if firstDownstream {
-                        uc.network.lock.Lock()
-                        seq, ok := uc.network.history[historyName]
-                        uc.network.lock.Unlock()
+                        net.lock.Lock()
+                        seq, ok := net.history[historyName]
+                        net.lock.Unlock()
                         if ok {
                                 seqPtr = &seq
@@ -715 +717 @@
 
                 // TODO: we need to create a consumer when adding networks on-the-fly
-                consumer, ch := uc.ring.NewConsumer(seqPtr)
+                consumer, ch := net.ring.NewConsumer(seqPtr)
                 go func() {
                         for {
@@ -721 +723 @@
                                 select {
                                 case <-ch:
+                                        uc := net.upstream()
+                                        if uc == nil {
+                                                dc.logger.Printf("ignoring messages for upstream %q: upstream is disconnected", net.Addr)
+                                                break
+                                        }
                                         dc.ringMessages <- ringMessage{consumer, uc}
                                 case <-dc.closed:
@@ -738 +745 @@
 
                         if lastDownstream {
-                                uc.network.lock.Lock()
-                                uc.network.history[historyName] = seq
-                                uc.network.lock.Unlock()
+                                net.lock.Lock()
+                                net.history[historyName] = seq
+                                net.lock.Unlock()
                         }
                 }()
@@ -1087 +1094 @@
                         dc.lock.Unlock()
 
-                        uc.ring.Produce(echoMsg)
+                        uc.network.ring.Produce(echoMsg)
                 }
         default: