Package org.apache.groovy.runtime.async

Class DefaultActor<T>

java.lang.Object
org.apache.groovy.runtime.async.DefaultActor<T>
Type Parameters:
T - the message type
All Implemented Interfaces:
Actor<T>, AutoCloseable
public final class DefaultActor<T> extends Object implements Actor<T>
Default implementation of Actor.

Each actor runs on a dedicated worker thread (virtual on JDK 21+) and dispatches messages sequentially from a FIFO mailbox, guaranteeing thread-safe state access without locks. The mailbox is a LinkedBlockingDeque so that ActorContext.unstashAll() can re-inject deferred messages at the head of the queue; a Semaphore gates send-side capacity for bounded mailboxes (the deque itself is internally unbounded). A message's permit is released as soon as the worker takes it from the deque — so the bound applies to the queue, not to messages currently being processed or held in the stash buffer; the stash buffer itself is unbounded.

Since:
6.0.0
See Also:

  • Method Details

    • currentlyExecuting

      public static Actor<?> currentlyExecuting()
      Implementation hook for Actor.currentSelf().

    • reactor

      public static <T, R> Actor<T> reactor(Function<T,R> handler, ActorOptions options)
      Description copied from interface: Actor
      Creates a stateless reactor actor with explicit ActorOptions controlling the mailbox and executor.

    • reactor

      public static <T, R> Actor<T> reactor(ReactorHandler<T,R> handler, ActorOptions options)
      Description copied from interface: Actor
      Context-aware variant of Actor.reactor(Function, ActorOptions).

    • stateful

      public static <T, S> Actor<T> stateful(S initialState, BiFunction<S,T,S> handler, ActorOptions options)
      Description copied from interface: Actor
      Creates a stateful actor with explicit ActorOptions controlling the mailbox and executor.

    • stateful

      public static <T, S> Actor<T> stateful(S initialState, StatefulHandler<S,T> handler, ActorOptions options)
      Description copied from interface: Actor
      Context-aware variant of Actor.stateful(Object, BiFunction, ActorOptions).

    • send

      public void send(T message)
      Description copied from interface: Actor
      Sends a message to this actor. The message is queued and processed asynchronously. Fire-and-forget — no reply is expected.

      Bounded-mailbox interaction (see ActorOptions.Overflow):

      • BLOCK: the calling thread blocks until queue capacity is available, then enqueues.
      • FAIL: throws IllegalStateException when the mailbox is full.
      • DROP_NEWEST: the message is silently dropped — there is no reply to carry the failure, so a fire-and-forget overflow is invisible to the sender. If you need drop visibility, prefer Actor.sendAndGet(T) (which binds the dropped reply to IllegalStateException) or a different overflow policy.
      Specified by:
      send in interface Actor<T>
      Parameters:
      message - the message to send

    • sendAndGet

      public <R> Awaitable<R> sendAndGet(T message)
      Description copied from interface: Actor
      Sends a message and returns an Awaitable that completes with the reply. For reactors, the reply is the handler's return value. For stateful actors, the reply is the new state.

      Bounded-mailbox interaction (see ActorOptions.Overflow):

      Specified by:
      sendAndGet in interface Actor<T>
      Type Parameters:
      R - the reply type
      Parameters:
      message - the message to send
      Returns:
      an awaitable reply

    • isActive

      public boolean isActive()
      Description copied from interface: Actor
      Returns true while the actor is accepting new sends.

      The actor lifecycle has three states, expressed via this method and Actor.isTerminated():

      • accepting — isActive() == true, isTerminated() == false: the actor accepts new sends and is processing them.
      • draining — isActive() == false, isTerminated() == false: entered immediately when Actor.stop() is called. Further sends throw IllegalStateException, but messages already queued (or sent in a race with stop) continue to run.
      • terminated — isActive() == false, isTerminated() == true: the worker has exited; the queue and any stash have been processed (or, for stashed sendAndGet replies, rejected).
      Specified by:
      isActive in interface Actor<T>

    • isTerminated

      public boolean isTerminated()
      Description copied from interface: Actor
      Returns true once the worker has fully exited — i.e. the queue and any stashed messages have been processed (or rejected, in the case of stashed sendAndGet replies). Always false while Actor.isActive() is true; becomes true some time after Actor.stop() is called, once draining completes.
      Specified by:
      isTerminated in interface Actor<T>

    • stop

      public void stop()
      Description copied from interface: Actor
      Stops this actor gracefully. Messages already in the queue are processed before the actor shuts down. New sends after stop throw IllegalStateException.
      Specified by:
      stop in interface Actor<T>

    • onError

      public Actor<T> onError(BiConsumer<Throwable,? super T> handler)
      Description copied from interface: Actor
      Registers a handler invoked when the message processor throws.

      Fire-and-forget Actor.send(T) otherwise has no way to surface a handler exception; this hook is the supported way to log, record metrics for, or react to those failures. For Actor.sendAndGet(T) the failure is still reported through the returned Awaitable; the onError handler runs in addition.

      To stop the actor from inside an error handler, prefer the context-aware overload Actor.onError(TriConsumer) — ctx.self().stop() works on any actor. Alternatively, if the actor was built with withCurrentSelf(true), call Actor.currentSelf().stop(); without that opt-in currentSelf() throws.

      Exceptions thrown from the handler itself are caught and discarded so the actor's processing loop is not destabilised. Replacing a previously registered handler replaces it wholesale — there is no chaining.

      Register before the first send. An onError call that happens after a message is already in flight may not see that message's failure: only handlers visible to the worker by the time it dispatches a given message are invoked for it. To guarantee coverage, chain onError into actor construction — as in the example below — and avoid registering it later. 

      
 def actor = Actor.reactor(handler).onError { Throwable t, msg ->
     log.warn("actor failed processing {}", msg, t)
 }
      Specified by:
      onError in interface Actor<T>
      Parameters:
      handler - invoked as (throwable, message)
      Returns:
      this actor, for chaining

    • onError

      public Actor<T> onError(TriConsumer<ActorContext<T>,Throwable,? super T> handler)
      Description copied from interface: Actor
      Context-aware variant of Actor.onError(BiConsumer). The handler receives an ActorContext that can be used to stop the actor via ctx.self().stop().
      Specified by:
      onError in interface Actor<T>
      Parameters:
      handler - invoked as (context, throwable, message)
      Returns:
      this actor, for chaining

    • toString

      public String toString()
      Overrides:
      toString in class Object