( cassette: CassetteService.Interface, name: string, request: WebSocketRequest, options: WebSocketRecorderOptions, redactor: Redactor, )
| 200 | }) |
| 201 | |
| 202 | const makeReplaySocket = ( |
| 203 | cassette: CassetteService.Interface, |
| 204 | name: string, |
| 205 | request: WebSocketRequest, |
| 206 | options: WebSocketRecorderOptions, |
| 207 | redactor: Redactor, |
| 208 | ): Effect.Effect<Socket.Socket, never, Scope.Scope> => |
| 209 | Effect.gen(function* () { |
| 210 | const replay = yield* makeReplayState(cassette, name, webSocketInteractions) |
| 211 | const active = yield* Ref.make<ActiveReplay | undefined>(undefined) |
| 212 | |
| 213 | return Socket.make({ |
| 214 | runRaw: (handler, runOptions) => |
| 215 | Effect.gen(function* () { |
| 216 | const claimed = yield* replay |
| 217 | .claim((interaction, index) => |
| 218 | Effect.sync(() => { |
| 219 | const incoming = openSnapshot(request, redactor) |
| 220 | if ( |
| 221 | interaction && |
| 222 | JSON.stringify(canonicalizeJson(incoming)) === JSON.stringify(canonicalizeJson(interaction.open)) |
| 223 | ) |
| 224 | return |
| 225 | throw new Error( |
| 226 | `WebSocket open ${index + 1}: expected ${safeText(interaction?.open)}, received ${safeText(incoming)}`, |
| 227 | ) |
| 228 | }), |
| 229 | ) |
| 230 | .pipe(Effect.orDie) |
| 231 | const progress = yield* Ref.make({ position: 0, changed: yield* Deferred.make<void>() }) |
| 232 | const writeLock = yield* Semaphore.make(1) |
| 233 | const state = { |
| 234 | interaction: claimed.interaction, |
| 235 | progress, |
| 236 | writeLock, |
| 237 | closed: yield* Ref.make(false), |
| 238 | } |
| 239 | const occupied = yield* Ref.modify(active, (current) => [current !== undefined, current ?? state]) |
| 240 | if (occupied) return yield* Effect.die("Concurrent runs of a replayed WebSocket are not supported") |
| 241 | yield* runReplay(state, handler, decodeEvent, runOptions?.onOpen).pipe( |
| 242 | Effect.ensuring(Ref.set(active, undefined)), |
| 243 | ) |
| 244 | }), |
| 245 | writer: Effect.succeed((message) => { |
| 246 | return Ref.get(active).pipe( |
| 247 | Effect.flatMap((state) => |
| 248 | state |
| 249 | ? state.writeLock.withPermit( |
| 250 | Effect.gen(function* () { |
| 251 | const current = yield* Ref.get(state.progress) |
| 252 | if (Socket.isCloseEvent(message)) { |
| 253 | yield* Ref.set(state.closed, true) |
| 254 | yield* Deferred.succeed(current.changed, undefined) |
| 255 | if (current.position === state.interaction.events.length) return |
| 256 | return yield* Effect.die( |
| 257 | new Error( |
| 258 | `WebSocket closed with unconsumed events: used ${current.position} of ${state.interaction.events.length}`, |
| 259 | ), |
no test coverage detected