| 109 | } |
| 110 | |
| 111 | func (rb *RingWriter[T, F]) Step() (normal bool) { |
| 112 | if !rb.disposeFlag.CompareAndSwap(0, 1) { |
| 113 | // already disposed |
| 114 | return |
| 115 | } |
| 116 | rb.LastValue = rb.Value |
| 117 | nextSeq := rb.LastValue.GetSequence() + 1 |
| 118 | next := rb.Next() |
| 119 | if normal = next.Value.StartWrite(); normal { |
| 120 | next.Value.Reset() |
| 121 | rb.Ring = next |
| 122 | } else { |
| 123 | rb.Reduce(1) //抛弃还有订阅者的节点 |
| 124 | rb.Ring = rb.Glow(1) //补充一个新节点 |
| 125 | if !rb.Value.StartWrite() { |
| 126 | panic("can't start write") |
| 127 | } |
| 128 | } |
| 129 | rb.Value.SetSequence(nextSeq) |
| 130 | rb.LastValue.Ready() |
| 131 | if !rb.disposeFlag.CompareAndSwap(1, 0) { |
| 132 | rb.Value.Ready() |
| 133 | } |
| 134 | return |
| 135 | } |
| 136 | |
| 137 | func (rb *RingWriter[T, F]) GetReaderCount() int32 { |
| 138 | return rb.ReaderCount.Load() |