update(Multicast): send tracked values to new subscriber first

multicast.go

@@ -115,30 +115,17 @@ func (m *multicast[T]) Emit(n Notification[T]) {
 func (m *multicast[T]) Subscribe(c Context, o Observer[T]) {
 	m.Mu.Lock()
 
-	lastn := m.LastN
-	if lastn.Kind == 0 {
-		c, o = Serialize(c, o)
-
-		o := o
-		m.Mobs.Add(&o)
-
-		c.AfterFunc(func() {
-			m.Mu.Lock()
-			m.Mobs.Delete(&o)
-			m.Mu.Unlock()
-			o.Error(c.Cause())
-		})
-	}
-
-	buf := m.Buf
-	if buf != nil {
+	if buf := m.Buf; buf != nil {
 		buf.RefCount.Add(1)
-		defer buf.RefCount.Add(^uint32(0))
-	}
+		decrease := true
+		defer func() {
+			if decrease {
+				buf.RefCount.Add(^uint32(0))
+			}
+		}()
 
-	m.Mu.Unlock()
+		m.Mu.Unlock()
 
-	if buf != nil {
 		q := buf.Queue
 		done := c.Done()
 
@@ -152,8 +139,30 @@ func (m *multicast[T]) Subscribe(c Context, o Observer[T]) {
 
 			Try1(o, Next(q.At(i)), func() { o.Error(ErrOops) })
 		}
+
+		buf.RefCount.Add(^uint32(0))
+		decrease = false
+
+		m.Mu.Lock()
 	}
 
+	lastn := m.LastN
+	if lastn.Kind == 0 {
+		c, o = Serialize(c, o)
+
+		o := o
+		m.Mobs.Add(&o)
+
+		c.AfterFunc(func() {
+			m.Mu.Lock()
+			m.Mobs.Delete(&o)
+			m.Mu.Unlock()
+			o.Error(c.Cause())
+		})
+	}
+
+	m.Mu.Unlock()
+
 	switch lastn.Kind {
 	case KindError:
 		o.Error(lastn.Error)