| 537 | } |
| 538 | |
| 539 | get upgrade(): Effect.Effect<Socket.Socket, Error.HttpServerError> { |
| 540 | return Effect.callback<Socket.Socket, Error.HttpServerError>((resume) => { |
| 541 | const deferred = Deferred.makeUnsafe<ServerWebSocket<WebSocketContext>>() |
| 542 | const closeDeferred = Deferred.makeUnsafe<void, Socket.SocketError>() |
| 543 | const semaphore = Semaphore.makeUnsafe(1) |
| 544 | |
| 545 | const success = this.bunServer.upgrade(this.source, { |
| 546 | data: { |
| 547 | deferred, |
| 548 | closeDeferred, |
| 549 | buffer: [], |
| 550 | run: wsDefaultRun |
| 551 | } |
| 552 | }) |
| 553 | if (!success) { |
| 554 | resume(Effect.fail( |
| 555 | new Error.HttpServerError({ |
| 556 | reason: new Error.RequestParseError({ |
| 557 | request: this, |
| 558 | description: "Not an upgradeable ServerRequest" |
| 559 | }) |
| 560 | }) |
| 561 | )) |
| 562 | return |
| 563 | } |
| 564 | resume(Effect.map(Deferred.await(deferred), (ws) => { |
| 565 | const write = (chunk: Uint8Array | string | Socket.CloseEvent) => |
| 566 | Effect.sync(() => { |
| 567 | if (typeof chunk === "string") { |
| 568 | ws.sendText(chunk) |
| 569 | } else if (Socket.isCloseEvent(chunk)) { |
| 570 | ws.close(chunk.code, chunk.reason) |
| 571 | } else { |
| 572 | ws.sendBinary(chunk) |
| 573 | } |
| 574 | |
| 575 | return true |
| 576 | }) |
| 577 | const writer = Effect.succeed(write) |
| 578 | const runRaw = Effect.fnUntraced( |
| 579 | function*<R, E, _>( |
| 580 | handler: (_: Uint8Array | string) => Effect.Effect<_, E, R> | void, |
| 581 | opts?: { readonly onOpen?: Effect.Effect<void> | undefined } |
| 582 | ) { |
| 583 | const set = yield* FiberSet.make<any, E>() |
| 584 | const run = yield* FiberSet.runtime(set)<R>() |
| 585 | function runRaw(data: Uint8Array | string) { |
| 586 | const result = handler(data) |
| 587 | if (Effect.isEffect(result)) { |
| 588 | run(result) |
| 589 | } |
| 590 | } |
| 591 | ws.data.run = runRaw |
| 592 | ws.data.buffer.forEach(runRaw) |
| 593 | ws.data.buffer.length = 0 |
| 594 | if (opts?.onOpen) yield* opts.onOpen |
| 595 | return yield* FiberSet.join(set) |
| 596 | }, |