feat: allow concur iterables as input to asAsync

src/operations/as.d.ts

@@ -19,6 +19,12 @@ import type { MaybePromiseLike } from '../internal/types.js'
 /**
  * Returns an async iterable wrapper around `iterable`.
  *
+ * Note that when passing a concur iterable the returned async iterable may have
+ * to buffer the values produced by the concur iterable because values may not
+ * be read from the async iterable as quickly as they are produced by the concur
+ * iterable. This is a fundamental problem because concur iterables are "push"
+ * based while async iterables are "pull" based, which creates backpressure.
+ *
  * @example
  * ```js
  * const asyncIterable = asAsync([`sloth`, `more sloth`, `even more sloth`])
@@ -35,7 +41,7 @@ import type { MaybePromiseLike } from '../internal/types.js'
  * ```
  */
 export const asAsync: <Value>(
-  iterable: Iterable<Value> | AsyncIterable<Value>,
+  iterable: Iterable<Value> | AsyncIterable<Value> | ConcurIterable<Value>,
 ) => AsyncIterable<Value>
 
 /**

src/operations/as.js

@@ -14,15 +14,48 @@
  * limitations under the License.
  */
 
-import { createAsyncIterable } from '../internal/helpers.js'
+import { createAsyncIterable, deferred } from '../internal/helpers.js'
 import { map } from './transform.js'
 
 export const asAsync = iterable => {
   if (iterable[Symbol.asyncIterator]) {
     return iterable
   }
 
-  return createAsyncIterable(() => iterable[Symbol.iterator]())
+  return createAsyncIterable(
+    iterable[Symbol.iterator]
+      ? () => iterable[Symbol.iterator]()
+      : async function* () {
+          let buffer = []
+          let done = false
+          let nonEmptyBufferDeferred = deferred()
+
+          iterable(value => {
+            buffer.push(value)
+            if (nonEmptyBufferDeferred) {
+              const currentDeferred = nonEmptyBufferDeferred
+              nonEmptyBufferDeferred = null
+              currentDeferred._resolve()
+            }
+          }).then(() => {
+            done = true
+            nonEmptyBufferDeferred?._resolve()
+          })
+
+          // eslint-disable-next-line no-unmodified-loop-condition
+          while (!done) {
+            if (!buffer.length) {
+              await nonEmptyBufferDeferred._promise
+              continue
+            }
+
+            const currentBuffer = buffer
+            buffer = []
+            nonEmptyBufferDeferred = deferred()
+            yield* currentBuffer
+          }
+        },
+  )
 }
 
 export const asConcur = iterable => {

test/operations/as.ts

@@ -40,10 +40,13 @@ test.skip(`asAsync types are correct`, () => {
   expectTypeOf(asAsync(asAsync([`a`, `b`, `c`]))).toMatchTypeOf<
     AsyncIterable<string>
   >()
+  expectTypeOf(asAsync(asConcur([`a`, `b`, `c`]))).toMatchTypeOf<
+    AsyncIterable<string>
+  >()
 })
 
 testProp(
-  `asAsync returns a pure async iterable`,
+  `asAsync returns a pure async iterable for non-concur iterables`,
   [fc.oneof(iterableArb, asyncIterableArb)],
   async ({ iterable }) => {
     const asyncIterable = asAsync(iterable)
@@ -53,7 +56,7 @@ testProp(
 )
 
 testProp(
-  `asAsync returns an async iterable containing the same values in the same order as the given iterable`,
+  `asAsync returns an async iterable containing the same values in the same order as the given iterable or async iterable`,
   [fc.oneof(iterableArb, asyncIterableArb)],
   async ({ iterable, values }) => {
     const asyncIterable = asAsync(iterable)
@@ -62,6 +65,18 @@ testProp(
   },
 )
 
+testProp(
+  `asAsync returns an async iterable containing the same values as the given concur iterable`,
+  [concurIterableArb],
+  async ({ iterable, values }) => {
+    const asyncIterable = asAsync(iterable)
+
+    expect(await reduceAsync(toArray(), asyncIterable)).toIncludeSameMembers(
+      values,
+    )
+  },
+)
+
 test.skip(`asConcur types are correct`, () => {
   expectTypeOf(asConcur([1, 2, 3])).toMatchTypeOf<ConcurIterable<number>>()
   expectTypeOf(asConcur([1, 2, 3] as Iterable<number>)).toMatchTypeOf<