add Atom tests

Author: tim-smart

packages/effect/src/Fiber.ts

@@ -72,7 +72,7 @@
  *
  * @since 2.0.0
  */
-import type { NonEmptyArray } from "./Array.ts"
+import type * as Arr from "./Array.ts"
 import type { Effect } from "./Effect.ts"
 import type { Exit } from "./Exit.ts"
 import * as effect from "./internal/effect.ts"
@@ -258,10 +258,13 @@ export const join: <A, E>(self: Fiber<A, E>) => Effect<A, E> = effect.fiberJoin
  * @since 2.0.0
  * @category combinators
  */
-export const joinAll: <A extends Fiber<any, any>>(
-  self: Iterable<A>
+export const joinAll: <A extends Iterable<Fiber<any, any>>>(
+  self: A
 ) => Effect<
-  NonEmptyArray<A extends Fiber<infer _A, infer _E> ? _A : never>,
+  Arr.ReadonlyArray.With<
+    A,
+    A extends Iterable<Fiber<infer _A, infer _E>> ? _A : never
+  >,
   A extends Fiber<infer _A, infer _E> ? _E : never
 > = effect.fiberJoinAll
 

packages/effect/src/Layer.ts

@@ -718,7 +718,7 @@ export const sync: {
  * @category constructors
  */
 export const syncServices = <A>(evaluate: LazyArg<ServiceMap.ServiceMap<A>>): Layer<A> =>
-  fromBuildUnsafe(constant(internalEffect.sync(evaluate)))
+  fromBuildMemo(constant(internalEffect.sync(evaluate)))
 
 /**
  * Constructs a layer from the specified scoped effect.

packages/effect/src/ServiceMap.ts

@@ -53,6 +53,7 @@ export interface Service<in out Identifier, in out Shape>
   of(self: Shape): Shape
   serviceMap(self: Shape): ServiceMap<Identifier>
   use<A, E, R>(f: (service: Shape) => Effect<A, E, R>): Effect<A, E, R | Identifier>
+  useSync<A>(f: (service: Shape) => A): Effect<A, never, Identifier>
 
   readonly stack?: string | undefined
   readonly key: string
@@ -205,6 +206,9 @@ const ServiceProto: any = {
   },
   use<A, E, R>(this: Service<never, any>, f: (service: any) => Effect<A, E, R>): Effect<A, E, R> {
     return withFiber((fiber) => f(get(fiber.services, this)))
+  },
+  useSync<A>(this: Service<never, any>, f: (service: any) => A): Effect<A, never, never> {
+    return withFiber((fiber) => exitSucceed(f(get(fiber.services, this))))
   }
 }
 

packages/effect/src/Stream.ts

@@ -1474,22 +1474,12 @@ export const tap: {
   options?: {
     readonly concurrency?: number | "unbounded" | undefined
   } | undefined
-): Stream<A, E | E2, R | R2> => {
-  const concurrency = options?.concurrency ?? 1
-  if (concurrency === 1 || concurrency === "unbounded") {
-    return self.channel.pipe(
-      Channel.tap(Effect.forEach(f, { discard: true, concurrency }), options),
-      fromChannel
-    )
-  }
-  return suspend(() => {
-    const withPermit = Effect.makeSemaphoreUnsafe(concurrency).withPermit
-    return self.channel.pipe(
-      Channel.tap(Effect.forEach((a) => withPermit(f(a)), { discard: true, concurrency }), options),
-      fromChannel
-    )
-  })
-})
+): Stream<A, E | E2, R | R2> =>
+  mapEffect(
+    self,
+    (a) => Effect.as(f(a), a),
+    options
+  ))
 
 /**
  * @since 2.0.0

packages/effect/src/internal/effect.ts

@@ -758,8 +758,8 @@ export const fiberJoin = <A, E>(self: Fiber.Fiber<A, E>): Effect.Effect<A, E> =>
 }
 
 /** @internal */
-export const fiberJoinAll = <A extends Fiber.Fiber<any, any>>(self: Iterable<A>): Effect.Effect<
-  Arr.NonEmptyArray<A extends Fiber.Fiber<infer _A, infer _E> ? _A : never>,
+export const fiberJoinAll = <A extends Iterable<Fiber.Fiber<any, any>>>(self: A): Effect.Effect<
+  Arr.ReadonlyArray.With<A, A extends Iterable<Fiber.Fiber<infer _A, infer _E>> ? _A : never>,
   A extends Fiber.Fiber<infer _A, infer _E> ? _E : never
 > =>
   callback((resume) => {

packages/effect/src/unstable/http/HttpBody.ts

@@ -214,15 +214,16 @@ export const text = (body: string, contentType?: string): Uint8Array =>
  * @since 4.0.0
  * @category constructors
  */
-export const jsonUnsafe = (body: unknown): Uint8Array => text(JSON.stringify(body), "application/json")
+export const jsonUnsafe = (body: unknown, contentType?: string): Uint8Array =>
+  text(JSON.stringify(body), contentType ?? "application/json")
 
 /**
  * @since 4.0.0
  * @category constructors
  */
-export const json = (body: unknown): Effect.Effect<Uint8Array, HttpBodyError> =>
+export const json = (body: unknown, contentType?: string): Effect.Effect<Uint8Array, HttpBodyError> =>
   Effect.try({
-    try: () => text(JSON.stringify(body), "application/json"),
+    try: () => text(JSON.stringify(body), contentType ?? "application/json"),
     catch: (cause) => new HttpBodyError({ reason: { _tag: "JsonError" }, cause })
   })
 
@@ -235,19 +236,19 @@ export const jsonSchema = <S extends Schema.Schema<any>>(
   options?: ParseOptions | undefined
 ) => {
   const encode = Parser.encodeUnknownEffect(Schema.toCodecJson(schema))
-  return (body: S["Type"]): Effect.Effect<Uint8Array, HttpBodyError, S["EncodingServices"]> =>
+  return (body: S["Type"], contentType?: string): Effect.Effect<Uint8Array, HttpBodyError, S["EncodingServices"]> =>
     encode(body, options).pipe(
       Effect.mapError((issue) => new HttpBodyError({ reason: { _tag: "SchemaError", issue }, cause: issue })),
-      Effect.flatMap((body) => json(body))
+      Effect.flatMap((body) => json(body, contentType))
     )
 }
 
 /**
  * @since 4.0.0
  * @category constructors
  */
-export const urlParams = (urlParams: UrlParams.UrlParams): Uint8Array =>
-  text(UrlParams.toString(urlParams), "application/x-www-form-urlencoded")
+export const urlParams = (urlParams: UrlParams.UrlParams, contentType?: string): Uint8Array =>
+  text(UrlParams.toString(urlParams), contentType ?? "application/x-www-form-urlencoded")
 
 /**
  * @since 4.0.0

packages/effect/src/unstable/http/HttpServerResponse.ts

@@ -92,7 +92,7 @@ export const empty = (
  */
 export const redirect = (
   location: string | URL,
-  options?: Options.WithContentType | undefined
+  options?: Options.WithContent | undefined
 ): HttpServerResponse => {
   const headers = Headers.fromRecordUnsafe({ location: location.toString() })
   return makeResponse({
@@ -213,16 +213,18 @@ export const htmlStream = <
  */
 export const json = (
   body: unknown,
-  options?: Options.WithContent | undefined
-): Effect.Effect<HttpServerResponse, Body.HttpBodyError> =>
-  Effect.map(Body.json(body), (body) =>
+  options?: Options.WithContentType | undefined
+): Effect.Effect<HttpServerResponse, Body.HttpBodyError> => {
+  const headers = options?.headers ? Headers.fromInput(options.headers) : Headers.empty
+  return Effect.map(Body.json(body, getContentType(options, headers)), (body) =>
     makeResponse({
       status: options?.status ?? 200,
       statusText: options?.statusText,
-      headers: options?.headers && Headers.fromInput(options.headers),
+      headers,
       cookies: options?.cookies,
       body
     }))
+}
 
 /**
  * @since 4.0.0
@@ -235,16 +237,18 @@ export const schemaJson = <A, I, RD, RE>(
   const encode = Body.jsonSchema(schema, options)
   return (
     body: A,
-    options?: Options.WithContent | undefined
-  ): Effect.Effect<HttpServerResponse, Body.HttpBodyError, RE> =>
-    Effect.map(encode(body), (body) =>
+    options?: Options.WithContentType | undefined
+  ): Effect.Effect<HttpServerResponse, Body.HttpBodyError, RE> => {
+    const headers = options?.headers ? Headers.fromInput(options.headers) : Headers.empty
+    return Effect.map(encode(body, getContentType(options, headers)), (body) =>
       makeResponse({
         status: options?.status ?? 200,
         statusText: options?.statusText,
-        headers: options?.headers && Headers.fromInput(options.headers),
+        headers,
         cookies: options?.cookies,
         body
       }))
+  }
 }
 
 /**
@@ -253,34 +257,38 @@ export const schemaJson = <A, I, RD, RE>(
  */
 export const jsonUnsafe = (
   body: unknown,
-  options?: Options.WithContent | undefined
-): HttpServerResponse =>
-  makeResponse({
+  options?: Options.WithContentType | undefined
+): HttpServerResponse => {
+  const headers = options?.headers ? Headers.fromInput(options.headers) : Headers.empty
+  return makeResponse({
     status: options?.status ?? 200,
     statusText: options?.statusText,
-    headers: options?.headers && Headers.fromInput(options.headers),
+    headers,
     cookies: options?.cookies,
-    body: Body.jsonUnsafe(body)
+    body: Body.jsonUnsafe(body, getContentType(options, headers))
   })
+}
 
 /**
  * @since 4.0.0
  * @category constructors
  */
 export const urlParams = (
   body: UrlParams.Input,
-  options?: Options.WithContent | undefined
-): HttpServerResponse =>
-  makeResponse({
+  options?: Options.WithContentType | undefined
+): HttpServerResponse => {
+  const headers = options?.headers ? Headers.fromInput(options.headers) : Headers.empty
+  return makeResponse({
     status: options?.status ?? 200,
     statusText: options?.statusText,
-    headers: options?.headers && Headers.fromInput(options.headers),
+    headers,
     cookies: options?.cookies,
     body: Body.text(
       UrlParams.toString(UrlParams.fromInput(body)),
-      "application/x-www-form-urlencoded"
+      getContentType(options, headers) ?? "application/x-www-form-urlencoded"
     )
   })
+}
 
 /**
  * @since 4.0.0

packages/effect/src/unstable/httpapi/HttpApiBuilder.ts

@@ -645,25 +645,33 @@ const responseTransformation = <A, I, RD, RE>(
       const encoding = HttpApiSchema.getEncoding(ast)
       switch (encoding.kind) {
         case "Json": {
-          return Effect.mapError(
-            Response.json(data, { status }),
-            (error) => new Issue.InvalidType(ast, Option.some(error))
-          )
+          try {
+            return Effect.succeed(Response.text(JSON.stringify(data), {
+              status,
+              contentType: encoding.contentType
+            }))
+          } catch (error) {
+            return Effect.fail(new Issue.InvalidType(ast, Option.some(error)))
+          }
         }
         case "Text": {
-          return Effect.succeed(Response.text(data as any, {
+          return Effect.succeed(Response.text(data as string, {
             status,
             contentType: encoding.contentType
           }))
         }
         case "Uint8Array": {
-          return Effect.succeed(Response.uint8Array(data as any, {
+          return Effect.succeed(Response.uint8Array(data as Uint8Array, {
             status,
             contentType: encoding.contentType
           }))
         }
         case "UrlParams": {
-          return Effect.succeed(Response.urlParams(data as any, { status }))
+          return Effect.succeed(
+            Response.urlParams(data as any, { status }).pipe(
+              Response.setHeader("content-type", encoding.contentType)
+            )
+          )
         }
       }
     }

packages/effect/src/unstable/reactivity/Atom.ts

@@ -9,7 +9,7 @@ import * as Effect from "../../Effect.ts"
 import * as Exit from "../../Exit.ts"
 import * as Fiber from "../../Fiber.ts"
 import type { LazyArg } from "../../Function.ts"
-import { constant, constVoid, dual, pipe } from "../../Function.ts"
+import { constant, constTrue, constVoid, dual, pipe } from "../../Function.ts"
 import type * as Inspectable from "../../Inspectable.ts"
 import { PipeInspectableProto } from "../../internal/core.ts"
 import * as Layer from "../../Layer.ts"
@@ -513,21 +513,20 @@ function runCallbackSync<R, A, E, ER = never>(
   }
   const runFork = Effect.runForkWith(services)
   const scheduler = ServiceMap.get(services, Scheduler.Scheduler)
-  const canFlush = "flush" in scheduler
-  const fiberRuntime = runFork(effect)
-  if (canFlush) {
+  const fiber = runFork(effect)
+  if ("flush" in scheduler) {
     ;(scheduler as Scheduler.MixedScheduler).flush()
   }
-  const result = fiberRuntime.pollUnsafe()
+  const result = fiber.pollUnsafe()
   if (result) {
     onExit(result)
     return undefined
   }
-  const remove = fiberRuntime.addObserver(onExit)
+  const remove = fiber.addObserver(onExit)
   function cancel() {
     remove()
     if (!uninterruptible) {
-      fiberRuntime.interruptUnsafe()
+      fiber.interruptUnsafe()
     }
   }
   return cancel
@@ -750,16 +749,16 @@ function makeStream<A, E>(
   services = ServiceMap.add(services, AtomRegistry, ctx.registry)
 
   const run = Effect.scopedWith((scope) =>
-    Effect.flatMap(Channel.toPullScoped(Stream.toChannel(stream), scope), (pull) =>
-      pull.pipe(
-        Effect.flatMap((arr) => {
+    Effect.flatMap(Channel.toPullScoped(stream.channel, scope), (pull) =>
+      Effect.whileLoop({
+        while: constTrue,
+        body: () => pull,
+        step(arr) {
           ctx.setSelf(AsyncResult.success(Arr.lastNonEmpty(arr), {
             waiting: true
           }))
-          return Effect.void
-        }),
-        Effect.forever({ autoYield: false })
-      ))
+        }
+      }))
   ).pipe(
     Effect.catchCause((cause) => {
       if (Pull.isHaltCause(cause)) {
@@ -784,8 +783,16 @@ function makeStream<A, E>(
       return Effect.void
     })
   )
+  const servicesMap = new Map(services.mapUnsafe)
+  servicesMap.set(AtomRegistry.key, ctx.registry)
+  servicesMap.set(Scheduler.Scheduler.key, ctx.registry.scheduler)
 
-  const cancel = runCallbackSync(ServiceMap.add(services, AtomRegistry, ctx.registry), run, constVoid, false)
+  const cancel = runCallbackSync(
+    ServiceMap.makeUnsafe<AtomRegistry>(servicesMap),
+    run,
+    constVoid,
+    false
+  )
   if (cancel !== undefined) {
     ctx.addFinalizer(cancel)
   }
@@ -1076,7 +1083,7 @@ function makeResultFn<Arg, E, A>(
           (fiber) => {
             fibers.add(fiber)
             fiber.addObserver(() => fibers.delete(fiber))
-            return Effect.map(Fiber.joinAll(fibers), Arr.lastNonEmpty)
+            return Effect.map(Fiber.joinAll(fibers), (arr) => arr[0])
           }
         )
       }