Add cross-target async compatibility runtime

.github/workflows/ci.yml

@@ -43,6 +43,18 @@ jobs:
         run: |
           moon check bitlogger_async --target native
 
+      - name: Check bitlogger_async wasm-gc
+        run: |
+          moon check bitlogger_async --target wasm-gc
+
+      - name: Check bitlogger_async js
+        run: |
+          moon check bitlogger_async --target js
+
+      - name: Test bitlogger_async wasm-gc
+        run: |
+          moon test bitlogger_async --target wasm-gc
+
       - name: Run basic example
         run: |
           moon run examples/basic

bitlogger_async/BitLoggerAsync_test.mbt

@@ -1,4 +1,5 @@
 async test "shutdown drains pending records" {
+  inspect(async_runtime_mode_label(async_runtime_mode()) == "native_worker" || async_runtime_mode_label(async_runtime_mode()) == "compatibility", content="true")
   let written : Ref[Array[String]] = Ref::new([])
   let flushes : Ref[Int] = Ref::new(0)
   let logger = async_logger(
@@ -148,3 +149,59 @@ test "async build config stringify roundtrips nested logger and async fields" {
     AsyncFlushPolicy::Shutdown => "Shutdown"
   }, content="Shutdown")
 }
+
+test "async runtime capability helpers stay consistent" {
+  let mode = async_runtime_mode()
+  let state = async_runtime_state()
+  let worker_supported = match mode {
+    AsyncRuntimeMode::NativeWorker => true
+    AsyncRuntimeMode::Compatibility => false
+  }
+  inspect(
+    async_runtime_mode_label(mode) == "native_worker" || async_runtime_mode_label(mode) == "compatibility",
+    content="true",
+  )
+  inspect(async_runtime_supports_background_worker() == worker_supported, content="true")
+  inspect(async_runtime_mode_label(state.mode) == async_runtime_mode_label(mode), content="true")
+  inspect(state.background_worker == worker_supported, content="true")
+  inspect(
+    stringify_async_runtime_state(state),
+    content=if worker_supported {
+      "{\"mode\":\"native_worker\",\"background_worker\":true}"
+    } else {
+      "{\"mode\":\"compatibility\",\"background_worker\":false}"
+    },
+  )
+}
+
+async test "run drains queued records in compatibility backends too" {
+  let written : Ref[Array[String]] = Ref::new([])
+  let logger = async_logger(
+    @bitlogger.callback_sink(fn(rec) {
+      written.val.push(rec.message)
+    }),
+    config=AsyncLoggerConfig::new(
+      max_pending=4,
+      overflow=AsyncOverflowPolicy::DropNewest,
+      max_batch=2,
+      linger_ms=5,
+      flush=AsyncFlushPolicy::Never,
+    ),
+    min_level=@bitlogger.Level::Info,
+    target="async.compat",
+  )
+
+  @async.with_task_group(group => {
+    logger.info("one")
+    logger.info("two")
+    inspect(logger.pending_count(), content="2")
+    group.spawn_bg(() => logger.run())
+    logger.shutdown()
+  })
+
+  inspect(logger.is_closed(), content="true")
+  inspect(logger.pending_count(), content="0")
+  inspect(written.val.length(), content="2")
+  inspect(written.val[0], content="one")
+  inspect(written.val[1], content="two")
+}

bitlogger_async/async_logger_native.mbt

@@ -14,6 +14,62 @@ pub(all) enum AsyncFlushPolicy {
   Shutdown
 }
 
+pub enum AsyncRuntimeMode {
+  NativeWorker
+  Compatibility
+}
+
+pub fn async_runtime_mode() -> AsyncRuntimeMode {
+  AsyncRuntimeMode::NativeWorker
+}
+
+pub fn async_runtime_mode_label(mode : AsyncRuntimeMode) -> String {
+  match mode {
+    AsyncRuntimeMode::NativeWorker => "native_worker"
+    AsyncRuntimeMode::Compatibility => "compatibility"
+  }
+}
+
+fn all_async_runtime_modes() -> Array[AsyncRuntimeMode] {
+  [AsyncRuntimeMode::NativeWorker, AsyncRuntimeMode::Compatibility]
+}
+
+pub fn async_runtime_supports_background_worker() -> Bool {
+  ignore(all_async_runtime_modes())
+  true
+}
+
+pub struct AsyncRuntimeState {
+  mode : AsyncRuntimeMode
+  background_worker : Bool
+}
+
+pub fn async_runtime_state() -> AsyncRuntimeState {
+  {
+    mode: async_runtime_mode(),
+    background_worker: async_runtime_supports_background_worker(),
+  }
+}
+
+pub fn async_runtime_state_to_json(state : AsyncRuntimeState) -> @json_parser.JsonValue {
+  @json_parser.JsonValue::Object({
+    "mode": @json_parser.JsonValue::String(async_runtime_mode_label(state.mode)),
+    "background_worker": @json_parser.JsonValue::Bool(state.background_worker),
+  })
+}
+
+pub fn stringify_async_runtime_state(
+  state : AsyncRuntimeState,
+  pretty~ : Bool = false,
+) -> String {
+  let value = async_runtime_state_to_json(state)
+  if pretty {
+    @json_parser.stringify_pretty(value, 2)
+  } else {
+    @json_parser.stringify(value)
+  }
+}
+
 pub struct AsyncLoggerConfig {
   max_pending : Int
   overflow : AsyncOverflowPolicy

bitlogger_async/async_logger_stub.mbt

@@ -14,6 +14,62 @@ pub(all) enum AsyncFlushPolicy {
   Shutdown
 }
 
+pub enum AsyncRuntimeMode {
+  NativeWorker
+  Compatibility
+}
+
+pub fn async_runtime_mode() -> AsyncRuntimeMode {
+  AsyncRuntimeMode::Compatibility
+}
+
+pub fn async_runtime_mode_label(mode : AsyncRuntimeMode) -> String {
+  match mode {
+    AsyncRuntimeMode::NativeWorker => "native_worker"
+    AsyncRuntimeMode::Compatibility => "compatibility"
+  }
+}
+
+fn all_async_runtime_modes() -> Array[AsyncRuntimeMode] {
+  [AsyncRuntimeMode::NativeWorker, AsyncRuntimeMode::Compatibility]
+}
+
+pub fn async_runtime_supports_background_worker() -> Bool {
+  ignore(all_async_runtime_modes())
+  false
+}
+
+pub struct AsyncRuntimeState {
+  mode : AsyncRuntimeMode
+  background_worker : Bool
+}
+
+pub fn async_runtime_state() -> AsyncRuntimeState {
+  {
+    mode: async_runtime_mode(),
+    background_worker: async_runtime_supports_background_worker(),
+  }
+}
+
+pub fn async_runtime_state_to_json(state : AsyncRuntimeState) -> @json_parser.JsonValue {
+  @json_parser.JsonValue::Object({
+    "mode": @json_parser.JsonValue::String(async_runtime_mode_label(state.mode)),
+    "background_worker": @json_parser.JsonValue::Bool(state.background_worker),
+  })
+}
+
+pub fn stringify_async_runtime_state(
+  state : AsyncRuntimeState,
+  pretty~ : Bool = false,
+) -> String {
+  let value = async_runtime_state_to_json(state)
+  if pretty {
+    @json_parser.stringify_pretty(value, 2)
+  } else {
+    @json_parser.stringify(value)
+  }
+}
+
 pub struct AsyncLoggerConfig {
   max_pending : Int
   overflow : AsyncOverflowPolicy
@@ -38,7 +94,102 @@ pub fn AsyncLoggerConfig::new(
   }
 }
 
-pub struct AsyncLogger[S] {}
+fn parse_async_overflow(name : String) -> AsyncOverflowPolicy raise {
+  match name.to_upper() {
+    "BLOCKING" => AsyncOverflowPolicy::Blocking
+    "DROPOLDEST" => AsyncOverflowPolicy::DropOldest
+    "DROPLATEST" => AsyncOverflowPolicy::DropNewest
+    "DROPNEWEST" => AsyncOverflowPolicy::DropNewest
+    _ => raise Failure::Failure("Unsupported async overflow policy: " + name)
+  }
+}
+
+fn parse_async_flush(name : String) -> AsyncFlushPolicy raise {
+  match name.to_upper() {
+    "NEVER" => AsyncFlushPolicy::Never
+    "NONE" => AsyncFlushPolicy::Never
+    "BATCH" => AsyncFlushPolicy::Batch
+    "SHUTDOWN" => AsyncFlushPolicy::Shutdown
+    _ => raise Failure::Failure("Unsupported async flush policy: " + name)
+  }
+}
+
+pub fn parse_async_logger_config_text(input : String) -> AsyncLoggerConfig raise {
+  let root = @json_parser.parse(input)
+  let obj = match root.as_object() {
+    Some(obj) => obj
+    None => raise Failure::Failure("Expected object for async logger config")
+  }
+  let max_pending = match obj.get("max_pending") {
+    Some(value) => match value.as_number() {
+      Some(number) => number.to_int()
+      None => raise Failure::Failure("Expected number at async_config.max_pending")
+    }
+    None => 0
+  }
+  let overflow = match obj.get("overflow") {
+    Some(value) => match value.as_string() {
+      Some(text) => parse_async_overflow(text)
+      None => raise Failure::Failure("Expected string at async_config.overflow")
+    }
+    None => AsyncOverflowPolicy::Blocking
+  }
+  let max_batch = match obj.get("max_batch") {
+    Some(value) => match value.as_number() {
+      Some(number) => number.to_int()
+      None => raise Failure::Failure("Expected number at async_config.max_batch")
+    }
+    None => 1
+  }
+  let linger_ms = match obj.get("linger_ms") {
+    Some(value) => match value.as_number() {
+      Some(number) => number.to_int()
+      None => raise Failure::Failure("Expected number at async_config.linger_ms")
+    }
+    None => 0
+  }
+  let flush = match obj.get("flush") {
+    Some(value) => match value.as_string() {
+      Some(text) => parse_async_flush(text)
+      None => raise Failure::Failure("Expected string at async_config.flush")
+    }
+    None => AsyncFlushPolicy::Never
+  }
+  AsyncLoggerConfig::new(
+    max_pending=max_pending,
+    overflow=overflow,
+    max_batch=max_batch,
+    linger_ms=linger_ms,
+    flush=flush,
+  )
+}
+
+pub fn async_logger_config_to_json(config : AsyncLoggerConfig) -> @json_parser.JsonValue {
+  @json_parser.JsonValue::Object({
+    "max_pending": @json_parser.JsonValue::Number(config.max_pending.to_double()),
+    "max_batch": @json_parser.JsonValue::Number(config.max_batch.to_double()),
+    "linger_ms": @json_parser.JsonValue::Number(config.linger_ms.to_double()),
+    "overflow": @json_parser.JsonValue::String(match config.overflow {
+      AsyncOverflowPolicy::Blocking => "Blocking"
+      AsyncOverflowPolicy::DropOldest => "DropOldest"
+      AsyncOverflowPolicy::DropNewest => "DropNewest"
+    }),
+    "flush": @json_parser.JsonValue::String(match config.flush {
+      AsyncFlushPolicy::Never => "Never"
+      AsyncFlushPolicy::Batch => "Batch"
+      AsyncFlushPolicy::Shutdown => "Shutdown"
+    }),
+  })
+}
+
+pub fn stringify_async_logger_config(config : AsyncLoggerConfig, pretty~ : Bool = false) -> String {
+  let value = async_logger_config_to_json(config)
+  if pretty {
+    @json_parser.stringify_pretty(value, 2)
+  } else {
+    @json_parser.stringify(value)
+  }
+}
 
 pub struct AsyncLoggerBuildConfig {
   logger : @bitlogger.LoggerConfig
@@ -53,182 +204,417 @@ pub fn AsyncLoggerBuildConfig::new(
 }
 
 pub fn parse_async_logger_build_config_text(input : String) -> AsyncLoggerBuildConfig raise {
-  ignore(input)
+  let root = @json_parser.parse(input)
-  abort("bitlogger_async currently only supports native/llvm backends")
+  let obj = match root.as_object() {
+    Some(obj) => obj
+    None => raise Failure::Failure("Expected object at async logger build config root")
   }
-
+  let logger = match obj.get("logger") {
-pub fn parse_async_logger_config_text(input : String) -> AsyncLoggerConfig raise {
+    Some(value) => @bitlogger.parse_logger_config_text(@json_parser.stringify(value))
-  ignore(input)
+    None => @bitlogger.default_logger_config()
-  abort("bitlogger_async currently only supports native/llvm backends")
   }
-
+  let async_config = match obj.get("async_config") {
-pub fn async_logger_config_to_json(config : AsyncLoggerConfig) -> @json_parser.JsonValue {
+    Some(value) => parse_async_logger_config_text(@json_parser.stringify(value))
-  ignore(config)
+    None => AsyncLoggerConfig::new()
-  abort("bitlogger_async currently only supports native/llvm backends")
   }
-
+  AsyncLoggerBuildConfig::new(logger=logger, async_config=async_config)
-pub fn stringify_async_logger_config(config : AsyncLoggerConfig, pretty~ : Bool = false) -> String {
-  ignore(config)
-  ignore(pretty)
-  abort("bitlogger_async currently only supports native/llvm backends")
 }
 
 pub fn async_logger_build_config_to_json(
   config : AsyncLoggerBuildConfig,
 ) -> @json_parser.JsonValue {
-  ignore(config)
+  @json_parser.JsonValue::Object({
-  abort("bitlogger_async currently only supports native/llvm backends")
+    "logger": @bitlogger.logger_config_to_json(config.logger),
+    "async_config": async_logger_config_to_json(config.async_config),
+  })
 }
 
 pub fn stringify_async_logger_build_config(
   config : AsyncLoggerBuildConfig,
   pretty~ : Bool = false,
 ) -> String {
-  ignore(config)
+  let value = async_logger_build_config_to_json(config)
-  ignore(pretty)
+  if pretty {
-  abort("bitlogger_async currently only supports native/llvm backends")
+    @json_parser.stringify_pretty(value, 2)
+  } else {
+    @json_parser.stringify(value)
+  }
 }
 
-pub fn async_logger[S : @bitlogger.Sink](
+pub struct AsyncLogger[S] {
+  min_level : @bitlogger.Level
+  target : String
+  timestamp : Bool
+  overflow : AsyncOverflowPolicy
+  max_batch : Int
+  linger_ms : Int
+  flush_policy : AsyncFlushPolicy
+  sink : S
+  flush_sink : (S) -> Int
+  context_fields : Array[@bitlogger.Field]
+  filter : (@bitlogger.Record) -> Bool
+  patch : @bitlogger.RecordPatch
+  queue : @async.Queue[@bitlogger.Record]
+  pending_count : Ref[Int]
+  dropped_count : Ref[Int]
+  is_closed : Ref[Bool]
+  is_running : Ref[Bool]
+  has_failed : Ref[Bool]
+  last_error : Ref[String]
+}
+
+pub fn[S] async_logger(
   sink : S,
   config~ : AsyncLoggerConfig = AsyncLoggerConfig::new(),
   min_level~ : @bitlogger.Level = @bitlogger.Level::Info,
   target~ : String = "",
   flush~ : (S) -> Int = fn(_) { 0 },
 ) -> AsyncLogger[S] {
-  ignore(sink)
+  {
-  ignore(config)
+    min_level,
-  ignore(min_level)
+    target,
-  ignore(target)
+    timestamp: false,
-  ignore(flush)
+    overflow: config.overflow,
-  abort("bitlogger_async currently only supports native/llvm backends")
+    max_batch: config.max_batch,
+    linger_ms: config.linger_ms,
+    flush_policy: config.flush,
+    sink,
+    flush_sink: flush,
+    context_fields: [],
+    filter: fn(_) { true },
+    patch: @bitlogger.identity_patch(),
+    queue: @async.Queue::new(kind=queue_kind_of(config)),
+    pending_count: Ref::new(0),
+    dropped_count: Ref::new(0),
+    is_closed: Ref::new(false),
+    is_running: Ref::new(false),
+    has_failed: Ref::new(false),
+    last_error: Ref::new(""),
+  }
+}
+
+fn queue_kind_of(config : AsyncLoggerConfig) -> @aqueue.Kind {
+  let limit = if config.max_pending < 0 { 0 } else { config.max_pending }
+  match config.overflow {
+    AsyncOverflowPolicy::Blocking => @aqueue.Kind::Blocking(limit)
+    AsyncOverflowPolicy::DropOldest => @aqueue.Kind::DiscardOldest(limit)
+    AsyncOverflowPolicy::DropNewest => @aqueue.Kind::DiscardLatest(limit)
+  }
 }
 
 pub fn[S] AsyncLogger::with_timestamp(self : AsyncLogger[S], enabled~ : Bool = true) -> AsyncLogger[S] {
-  ignore(self)
+  { ..self, timestamp: enabled }
-  ignore(enabled)
-  abort("bitlogger_async currently only supports native/llvm backends")
 }
 
 pub fn[S] AsyncLogger::with_target(self : AsyncLogger[S], target : String) -> AsyncLogger[S] {
-  ignore(self)
+  { ..self, target }
-  ignore(target)
-  abort("bitlogger_async currently only supports native/llvm backends")
 }
 
 pub fn[S] AsyncLogger::with_context_fields(
   self : AsyncLogger[S],
   fields : Array[@bitlogger.Field],
 ) -> AsyncLogger[S] {
-  ignore(self)
+  { ..self, context_fields: fields }
-  ignore(fields)
-  abort("bitlogger_async currently only supports native/llvm backends")
 }
 
 pub fn[S] AsyncLogger::with_filter(
   self : AsyncLogger[S],
   predicate : (@bitlogger.Record) -> Bool,
 ) -> AsyncLogger[S] {
-  ignore(self)
+  let current = self.filter
-  ignore(predicate)
+  {
-  abort("bitlogger_async currently only supports native/llvm backends")
+    ..self,
+    filter: fn(rec) {
+      current(rec) && predicate(rec)
+    },
+  }
 }
 
 pub fn[S] AsyncLogger::with_patch(
   self : AsyncLogger[S],
   patch : @bitlogger.RecordPatch,
 ) -> AsyncLogger[S] {
-  ignore(self)
+  let current = self.patch
-  ignore(patch)
+  {
-  abort("bitlogger_async currently only supports native/llvm backends")
+    ..self,
+    patch: fn(rec) {
+      patch(current(rec))
+    },
+  }
 }
 
 pub fn[S] AsyncLogger::with_min_level(
   self : AsyncLogger[S],
   min_level : @bitlogger.Level,
 ) -> AsyncLogger[S] {
-  ignore(self)
+  { ..self, min_level }
-  ignore(min_level)
+}
-  abort("bitlogger_async currently only supports native/llvm backends")
+
+fn combine_targets(parent : String, child : String) -> String {
+  if parent == "" {
+    child
+  } else if child == "" {
+    parent
+  } else {
+    "\{parent}.\{child}"
+  }
 }
 
 pub fn[S] AsyncLogger::child(self : AsyncLogger[S], target : String) -> AsyncLogger[S] {
-  ignore(self)
+  { ..self, target: combine_targets(self.target, target) }
-  ignore(target)
-  abort("bitlogger_async currently only supports native/llvm backends")
 }
 
 pub fn[S] AsyncLogger::is_enabled(self : AsyncLogger[S], level : @bitlogger.Level) -> Bool {
-  ignore(self)
+  level.enabled(self.min_level)
-  ignore(level)
+}
-  abort("bitlogger_async currently only supports native/llvm backends")
+
+fn merge_fields(
+  left : Array[@bitlogger.Field],
+  right : Array[@bitlogger.Field],
+) -> Array[@bitlogger.Field] {
+  if left.length() == 0 {
+    right
+  } else if right.length() == 0 {
+    left
+  } else {
+    left + right
+  }
+}
+
+pub async fn[S] AsyncLogger::log(
+  self : AsyncLogger[S],
+  level : @bitlogger.Level,
+  message : String,
+  fields~ : Array[@bitlogger.Field] = [],
+  target? : String = "",
+) -> Unit {
+  guard !self.is_closed() else {
+    ()
+  }
+  guard self.is_enabled(level) else {
+    ()
+  }
+  let actual_target = if target == "" { self.target } else { target }
+  let timestamp_ms = if self.timestamp { @env.now() } else { 0UL }
+  let rec = @bitlogger.Record::new(
+    level,
+    message,
+    timestamp_ms=timestamp_ms,
+    target=actual_target,
+    fields=merge_fields(self.context_fields, fields),
+  )
+  let rec = (self.patch)(rec)
+  guard (self.filter)(rec) else {
+    ()
+  }
+  let accepted = self.queue.try_put(rec) catch {
+    err if err is AsyncLoggerClosed => false
+    err => raise err
+  }
+  if accepted {
+    self.pending_count.val += 1
+  } else {
+    match self.overflow {
+      AsyncOverflowPolicy::Blocking => {
+        self.queue.put(rec) catch {
+          err if err is AsyncLoggerClosed => ()
+          err => raise err
+        }
+        self.pending_count.val += 1
+      }
+      AsyncOverflowPolicy::DropOldest | AsyncOverflowPolicy::DropNewest => {
+        self.dropped_count.val += 1
+      }
+    }
+  }
+}
+
+pub async fn[S] AsyncLogger::trace(
+  self : AsyncLogger[S],
+  message : String,
+  fields~ : Array[@bitlogger.Field] = [],
+) -> Unit {
+  self.log(@bitlogger.Level::Trace, message, fields=fields)
+}
+
+pub async fn[S] AsyncLogger::debug(
+  self : AsyncLogger[S],
+  message : String,
+  fields~ : Array[@bitlogger.Field] = [],
+) -> Unit {
+  self.log(@bitlogger.Level::Debug, message, fields=fields)
+}
+
+pub async fn[S] AsyncLogger::info(
+  self : AsyncLogger[S],
+  message : String,
+  fields~ : Array[@bitlogger.Field] = [],
+) -> Unit {
+  self.log(@bitlogger.Level::Info, message, fields=fields)
+}
+
+pub async fn[S] AsyncLogger::warn(
+  self : AsyncLogger[S],
+  message : String,
+  fields~ : Array[@bitlogger.Field] = [],
+) -> Unit {
+  self.log(@bitlogger.Level::Warn, message, fields=fields)
+}
+
+pub async fn[S] AsyncLogger::error(
+  self : AsyncLogger[S],
+  message : String,
+  fields~ : Array[@bitlogger.Field] = [],
+) -> Unit {
+  self.log(@bitlogger.Level::Error, message, fields=fields)
 }
 
 pub fn[S] AsyncLogger::pending_count(self : AsyncLogger[S]) -> Int {
-  ignore(self)
+  self.pending_count.val
-  abort("bitlogger_async currently only supports native/llvm backends")
 }
 
 pub fn[S] AsyncLogger::dropped_count(self : AsyncLogger[S]) -> Int {
-  ignore(self)
+  self.dropped_count.val
-  abort("bitlogger_async currently only supports native/llvm backends")
 }
 
 pub fn[S] AsyncLogger::is_closed(self : AsyncLogger[S]) -> Bool {
-  ignore(self)
+  self.is_closed.val
-  abort("bitlogger_async currently only supports native/llvm backends")
 }
 
 pub fn[S] AsyncLogger::is_running(self : AsyncLogger[S]) -> Bool {
-  ignore(self)
+  self.is_running.val
-  abort("bitlogger_async currently only supports native/llvm backends")
 }
 
 pub fn[S] AsyncLogger::has_failed(self : AsyncLogger[S]) -> Bool {
-  ignore(self)
+  self.has_failed.val
-  abort("bitlogger_async currently only supports native/llvm backends")
 }
 
 pub fn[S] AsyncLogger::last_error(self : AsyncLogger[S]) -> String {
-  ignore(self)
+  self.last_error.val
-  abort("bitlogger_async currently only supports native/llvm backends")
 }
 
 pub fn[S] AsyncLogger::flush_policy(self : AsyncLogger[S]) -> AsyncFlushPolicy {
-  ignore(self)
+  self.flush_policy
-  abort("bitlogger_async currently only supports native/llvm backends")
 }
 
 pub fn[S] AsyncLogger::close(self : AsyncLogger[S], clear? : Bool = false) -> Unit {
-  ignore(self)
+  self.is_closed.val = true
-  ignore(clear)
+  if clear {
-  abort("bitlogger_async currently only supports native/llvm backends")
+    let abandoned = self.pending_count()
+    if abandoned > 0 {
+      self.dropped_count.val += abandoned
+      self.pending_count.val = 0
+    }
+  }
+  self.queue.close(error=AsyncLoggerClosed, clear=clear)
 }
 
 pub async fn[S] AsyncLogger::wait_idle(self : AsyncLogger[S]) -> Unit {
-  ignore(self)
+  while self.pending_count() > 0 {
-  abort("bitlogger_async currently only supports native/llvm backends")
+    if self.has_failed() {
+      break
+    }
+    @async.pause()
+  }
 }
 
 pub async fn[S] AsyncLogger::shutdown(self : AsyncLogger[S], clear? : Bool = false) -> Unit {
-  ignore(self)
+  if clear {
-  ignore(clear)
+    self.close(clear=true)
-  abort("bitlogger_async currently only supports native/llvm backends")
+  } else {
+    self.wait_idle()
+    self.close()
+  }
 }
 
 pub async fn[S : @bitlogger.Sink] AsyncLogger::run(self : AsyncLogger[S]) -> Unit {
-  ignore(self)
+  self.is_running.val = true
-  abort("bitlogger_async currently only supports native/llvm backends")
+  self.has_failed.val = false
+  self.last_error.val = ""
+  run_worker(self) catch {
+    err => {
+      self.has_failed.val = true
+      self.last_error.val = err.to_string()
+      self.is_running.val = false
+      raise err
+    }
+  }
+  self.is_running.val = false
+}
+
+async fn[S : @bitlogger.Sink] run_worker(logger : AsyncLogger[S]) -> Unit {
+  while true {
+    let rec = logger.queue.get() catch {
+      err if err is AsyncLoggerClosed => break
+      err => raise err
+    }
+    logger.sink.write(rec)
+    if logger.pending_count.val > 0 {
+      logger.pending_count.val -= 1
+    }
+    for drained = 1; drained < logger.max_batch; {
+      let next = logger.queue.try_get() catch {
+        err if err is AsyncLoggerClosed => None
+        err => raise err
+      }
+      match next {
+        Some(next) => {
+          logger.sink.write(next)
+          if logger.pending_count.val > 0 {
+            logger.pending_count.val -= 1
+          }
+          continue drained + 1
+        }
+        None => {
+          if logger.linger_ms <= 0 {
+            break
+          }
+          let waited = @async.with_timeout_opt(logger.linger_ms, () => logger.queue.get()) catch {
+            err if err is AsyncLoggerClosed => None
+            err => raise err
+          }
+          match waited {
+            Some(next) => {
+              logger.sink.write(next)
+              if logger.pending_count.val > 0 {
+                logger.pending_count.val -= 1
+              }
+              continue drained + 1
+            }
+            None => break
+          }
+        }
+      }
+    }
+    match logger.flush_policy {
+      AsyncFlushPolicy::Batch => ignore((logger.flush_sink)(logger.sink))
+      _ => ()
+    }
+  }
+  match logger.flush_policy {
+    AsyncFlushPolicy::Shutdown => ignore((logger.flush_sink)(logger.sink))
+    _ => ()
+  }
 }
 
 pub fn build_async_logger(
   config : AsyncLoggerBuildConfig,
 ) -> AsyncLogger[@bitlogger.RuntimeSink] {
-  ignore(config)
+  let logger = @bitlogger.build_logger(config.logger)
-  abort("bitlogger_async currently only supports native/llvm backends")
+  async_logger(
+    logger.sink,
+    config=config.async_config,
+    min_level=logger.min_level,
+    target=logger.target,
+    flush=fn(sink) { sink.flush() },
+  ).with_timestamp(enabled=logger.timestamp)
 }
 
 pub fn build_async_text_logger(config : AsyncLoggerBuildConfig) -> AsyncLogger[@bitlogger.FormattedConsoleSink] {
-  ignore(config)
+  async_logger(
-  abort("bitlogger_async currently only supports native/llvm backends")
+    @bitlogger.text_console_sink(config.logger.sink.text_formatter.to_formatter()),
+    config=config.async_config,
+    min_level=config.logger.min_level,
+    target=config.logger.target,
+  ).with_timestamp(enabled=config.logger.timestamp)
 }

bitlogger_async/moon.pkg

@@ -7,8 +7,6 @@ import {
   "moonbitlang/core/ref",
 }
 
-supported_targets = "+native"
-
 options(
   targets: {
     "async_logger_native.mbt": [ "native", "llvm" ],