(self: Subscription<A>)
| 1142 | * @since 4.0.0 |
| 1143 | */ |
| 1144 | export const take = <A>(self: Subscription<A>): Effect.Effect<A> => |
| 1145 | Effect.suspend(() => { |
| 1146 | if (self.shutdownFlag.current) { |
| 1147 | return Effect.interrupt |
| 1148 | } |
| 1149 | if (self.replayWindow.remaining > 0) { |
| 1150 | const message = self.replayWindow.take()! |
| 1151 | return Effect.succeed(message) |
| 1152 | } |
| 1153 | const message = self.pollers.length === 0 |
| 1154 | ? self.subscription.poll() |
| 1155 | : MutableList.Empty |
| 1156 | if (message === MutableList.Empty) { |
| 1157 | return pollForItem(self) |
| 1158 | } else { |
| 1159 | self.strategy.onPubSubEmptySpaceUnsafe(self.pubsub, self.subscribers) |
| 1160 | return Effect.succeed(message) |
| 1161 | } |
| 1162 | }) |
| 1163 | |
| 1164 | /** |
| 1165 | * Takes all available messages from the subscription, suspending if no items |
no test coverage detected