val result: Seq[Int] = (1 to 5).map(_ => performAsyncIO)
                             // ^^^ map is the regular Seq.map implementation!

Mid-level, unstructured asynchronous operations

• Sources are the abstractions of asynchronous resources that an Async context can .awaitResult for. From the high-level interface, we have been using Future[T], which is actually a Source[Try[T]] that has a special property: once completed, Future[T] always return the same result.
• Listeners are the primary out-of-band way to receive values from a Source. The most simple implementation of a listener can be done by Listener.apply, which takes the item T and the origin Source[T] and do something with it! Listeners always receive at most one item from the Source, and are removed from the Source's list once completed. Listener bodies are run on the same computation that resolves the sources, so usage of them needs to be carefully tuned for a high degree of concurrency performance.
  • Locking listeners adds synchronization capabilities to a listener, to allow it to listen to multiple sources and decide whether it would be open to accepting the item or it is already expired. To know more, check out the Listener interface.
• Channels are bidirectional channels that can be use as a communication mean between concurrent processes. They provide .send(x: T)(using Async) and .read()(using Async) as suspending methods, but also .readSource and .sendSource(x: T) sources that reads/sends an item when a listener is attached and accepts the item/event. Three types of channels are provided:
  • SyncChannels are unbuffered synchronous (rendezvous) channels where sends and reads block until the transfer is actually made.
  • BufferedChannels allow sends to be buffered up to a certain buffer count, completing instantly if buffer space is available.
  • UnboundedChannels allow sends to be always buffered with a growable buffer, returning instantly in all cases. It exposes a .sendImmediately method that allows sending without an Async context.
• race, select and channel synchronization: gears provide exclusivity when using the above Async.select method with sources: exactly one of the given sources given will be resolved by the select (and no other item/event will be consumed from other sources). This is especially important for channels, as you typically would not want items from a channel being raced and thrown away:

  Async.blocking:
    val c1 = UnboundedChannel[Int]()
    val c2 = UnboundedChannel[Int]()
    c1.sendImmediately(1)
    c2.sendImmediately(2)
  
    def getInt = Async.select(
      c1.readSource.andThen(_ + 1),
      c2.readSource.andThen(_ + 2),
    )
  
    val ints = Seq(getInt, getInt) // Seq(1, 2) or Seq(2, 1)
  

  race has a similar exclusivity semantics to select, but expects a list of Source[T] as parameters and return a Source[T], resolving with an item when one of the inputs produces one. This allows you to compose Sources by racing them losslessly:

  val c1 = UnboundedChannel[Int]()
  val c2 = UnboundedChannel[Int]()
  val c3 = UnboundedChannel[Int]()
  c1.sendImmediately(1)
  c2.sendImmediately(2)
  
  val cs12: Source[Int] = Async.race(c1.readSource, c2.readSource)
  val cs123: Source[Int] = Async.race(cs12, c3.readSource)
  
  val ints = Async.blocking { Seq(cs12.awaitResult, cs123.awaitResult) } // Seq(1, 2) or Seq(2, 1)
  

• Creating sources: Source is a trait that can be manually implemented, but some Source creation methods are available:
  • With Promise and Future.withResolver: Promise is a wrapper for a Future that gets externally resolved. It exposes a .complete method which can be called externally to resolve the future and continue all awaiting computations. Future.withResolver exposes the same mechanics, but in a more familiar interface to Node.js's Promise constructor, while also allowing the caller to register a method that cleans up the external process when the Future is cancelled.
  • With Source.values: creates a very simple queue-like source that resolves items to listeners as they come.
  • Source transformation API: Source[T] exposes .transformValuesWith which applies a transformation function to every item in the original source before passing on a listener. This can be used to transform sources, but is not recommended, as the transformation will be run on the same computation as the original source and might cause unexpected performance problems.

Low-level suspension API and scheduling

gears's Async context requires the machinery of a Suspension API, which resembles "delimited continuations" to work. While default implementations for JVM 21+ and Scala Native 0.5+ exists, it is entirely replaceable and custom implementations (for example, to provide compatibility with JVM <= 20 with custom fibers) can be provided to create an Async context through Async.blocking. A Scheduler is also needed. By default, the default JVM virtual thread scheduler is used on JVM and ForkJoinPool is used on Scala Native.

Some "primitive" operations, such as sleep, are required from the scheduler/suspension implementation through the AsyncOperations interface. Again, default implementations are provided, but this can be customized.

What's coming up next?

• A strawman interface for cross platform IO: A separate library gears-io is in the early stage of design, which will provide a common asynchronous I/O (file, sockets) interface using Gear's Async context functions. We hope to enable cross-platform performant asynchronous I/O through this interface.
• More refinement to the API and support for structured concurrency
• Your input! We would appreciate any effort to start using the library, build something cool and report the experience!