92 lines
3.4 KiB
TypeScript
92 lines
3.4 KiB
TypeScript
import { assert, describe, it } from "@effect/vitest"
|
|
import { type Cause, Effect, Queue, Schema, Stream } from "effect"
|
|
import { Entity, ShardingConfig } from "effect/unstable/cluster"
|
|
import { Rpc } from "effect/unstable/rpc/index"
|
|
import { CallerId, ContextBleedEntity, ContextBleedLayer, TestEntity, TestEntityLayer, User } from "./TestEntity.ts"
|
|
|
|
const StreamEntity = Entity.make("StreamEntity", [
|
|
Rpc.make("Watch", {
|
|
success: Schema.Number,
|
|
stream: true
|
|
})
|
|
])
|
|
|
|
describe.concurrent("Entity", () => {
|
|
describe("makeTestClient", () => {
|
|
it.effect("creates an in-memory client for an entity layer", () =>
|
|
Effect.gen(function*() {
|
|
const makeClient = yield* Entity.makeTestClient(TestEntity, TestEntityLayer)
|
|
const client = yield* makeClient("123")
|
|
const user = yield* client.GetUser({ id: 1 })
|
|
assert.deepEqual(user, new User({ id: 1, name: "User 1" }))
|
|
}).pipe(Effect.provide(TestShardingConfig)))
|
|
|
|
it.effect("does not freeze the acquiring fiber's context into the entity server", () =>
|
|
Effect.gen(function*() {
|
|
const makeClient = yield* Entity.makeTestClient(ContextBleedEntity, ContextBleedLayer)
|
|
|
|
const client = yield* makeClient("1").pipe(Effect.provideService(CallerId, "A"))
|
|
|
|
const observed = yield* client.ReadCaller()
|
|
assert.strictEqual(observed, "none")
|
|
}).pipe(Effect.provide(TestShardingConfig)))
|
|
})
|
|
|
|
describe("toLayerQueue", () => {
|
|
it.effect("replies to a streaming RPC with a Stream", () =>
|
|
Effect.gen(function*() {
|
|
const layer = StreamEntity.toLayerQueue((mailbox, replier) =>
|
|
Effect.gen(function*() {
|
|
while (true) {
|
|
const req = yield* Queue.take(mailbox)
|
|
yield* replier.succeed(req, Stream.make(1, 2, 3))
|
|
}
|
|
})
|
|
)
|
|
|
|
const makeClient = yield* Entity.makeTestClient(StreamEntity, layer)
|
|
const client = yield* makeClient("entity-1")
|
|
const results: Array<number> = []
|
|
yield* client.Watch().pipe(
|
|
Stream.take(3),
|
|
Stream.runForEach((n) => Effect.sync(() => results.push(n)))
|
|
)
|
|
assert.deepEqual(results, [1, 2, 3])
|
|
}).pipe(Effect.provide(TestShardingConfig)))
|
|
|
|
it.effect("replies to a streaming RPC with a Dequeue", () =>
|
|
Effect.gen(function*() {
|
|
const layer = StreamEntity.toLayerQueue((mailbox, replier) =>
|
|
Effect.gen(function*() {
|
|
while (true) {
|
|
const req = yield* Queue.take(mailbox)
|
|
const q = yield* Queue.make<number, Cause.Done>()
|
|
yield* replier.succeed(req, q)
|
|
yield* Queue.offer(q, 1)
|
|
yield* Queue.offer(q, 2)
|
|
yield* Queue.offer(q, 3)
|
|
yield* Queue.end(q)
|
|
}
|
|
}) as Effect.Effect<never>
|
|
)
|
|
|
|
const makeClient = yield* Entity.makeTestClient(StreamEntity, layer)
|
|
const client = yield* makeClient("entity-1")
|
|
const results: Array<number> = []
|
|
yield* client.Watch().pipe(
|
|
Stream.take(3),
|
|
Stream.runForEach((n) => Effect.sync(() => results.push(n)))
|
|
)
|
|
assert.deepEqual(results, [1, 2, 3])
|
|
}).pipe(Effect.provide(TestShardingConfig)))
|
|
})
|
|
})
|
|
|
|
const TestShardingConfig = ShardingConfig.layer({
|
|
shardsPerGroup: 300,
|
|
entityMailboxCapacity: 10,
|
|
entityTerminationTimeout: 0,
|
|
entityMessagePollInterval: 5000,
|
|
sendRetryInterval: 100
|
|
})
|