Jetlang

Jetlang is a high-performance Java threading library designed for message-based concurrency, similar to the event-based actor model in Scala. It is intended as a complement to java.util.concurrent for high-performance, in-memory messaging.

Key characteristics include:

• Sequential Message Delivery: All messages to a specific Fiber are delivered sequentially, allowing components to maintain state without manual synchronization or thread race concerns.
• Low Lock Contention: Optimized to avoid central bottlenecks (like a single lock on a central thread pool), allowing performance to scale with application needs.
• Thread-Safe Publishing: The publish mechanism is thread-safe, making it easy to integrate with other threading models.
• Flexible Backing: A single Fiber interface can be backed by either a dedicated thread or a thread pool.
• No Dependencies: The library requires only the JDK (1.6+).

Jetlang provides Channel objects to facilitate message passing between different threads or fibers. This allows for decoupled, asynchronous communication.

Workflow:

1. Create a Fiber to act as the receiver and call .start().
2. Create a Channel<T> (e.g., MemoryChannel<T>).
3. Define a Callback<T> to handle incoming messages.
4. Subscribe the receiver fiber to the channel using channel.subscribe(receiver, callback).
5. Publish messages using channel.publish(message). This method is thread-safe.
6. Call .dispose() on the receiver fiber when finished.

// start thread backed receiver. 
// Lighweight fibers can also be created using a thread pool
Fiber receiver = new ThreadFiber();
receiver.start();

// create java.util.concurrent.CountDownLatch to notify when message arrives
final CountDownLatch latch = new CountDownLatch(1);

// create channel to message between threads
Channel<String> channel = new MemoryChannel<String>();

Callback<String> onMsg = new Callback<String>() {
    public void onMessage(String message) {
        //open latch
        latch.countDown();
    }
};
//add subscription for message on receiver thread
channel.subscribe(receiver, onMsg);

//publish message to receive thread. the publish method is thread safe.
channel.publish("Hello");

//wait for receiving thread to receive message
latch.await(10, TimeUnit.SECONDS);

//shutdown thread
receiver.dispose();

A PoolFiber allows you to schedule commands to run after a delay or at fixed intervals. These scheduled tasks return a Disposable object which can be used to cancel the task.

Available scheduling methods:

• schedule(Runnable command, long delay, TimeUnit unit): Runs the command once after the specified delay.
• scheduleAtFixedRate(Runnable command, long initialDelay, long delay, TimeUnit unit): Runs the command repeatedly at a fixed rate.
• scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit): Runs the command repeatedly with a fixed delay between the end of one execution and the start of the next.

Note: When you call these methods, the returned Disposable is automatically registered with the fiber. When the fiber is dispose()'d, all scheduled tasks are also stopped.

Disposable task = fiber.schedule(() -> {
    System.out.println("Delayed task");
}, 5, TimeUnit.SECONDS);

// To cancel manually:
task.dispose();

When subscribing to a MemoryRequestChannel, you can provide an additional callback to handle SessionClosed<R> events. This is useful for detecting when a request has been disposed or completed, allowing you to clean up resources associated with that specific request session.

To use this, call the subscribe method that accepts three arguments: DisposingExecutor, Callback<Request<R, V>> (for the request itself), and Callback<SessionClosed<R>> (for the closure event).

channel.subscribe(executor, 
    new Callback<Request<String, Integer>>() {
        @Override
        public void onMessage(Request<String, Integer> request) {
            // Process request
        }
    }, 
    new Callback<SessionClosed<String>>() {
        @Override
        public void onMessage(SessionClosed<String> closed) {
            String originalRequest = closed.getOriginalRequest();
            Session session = closed.getSession();
            // Handle session cleanup
        }
    }
);

A PoolFiber is a Fiber implementation that uses a provided Executor (thread pool) to process commands. It manages a queue of tasks and uses a BatchExecutor to execute them. It also supports scheduling tasks with delays or fixed rates via an internal scheduler.

To instantiate a PoolFiber, you must provide:

1. An Executor (the thread pool for flushing tasks).
2. A BatchExecutor (to execute the command buffers).
3. A ScheduledExecutorService (for time-based scheduling).

Key lifecycle methods:

• start(): Transitions the fiber to the Running state and begins processing the queue.
• execute(Runnable commands): Adds commands to the fiber's queue for execution.
• dispose(): Stops the fiber, prevents new commands from being queued, and disposes of all registered Disposable objects.

PoolFiber fiber = new PoolFiber(pool, batchExecutor, scheduler);
fiber.start();
fiber.execute(() -> {
    // Your commands here
});
// ...
fiber.dispose();

MemoryRequestChannel<R, V> is an implementation of RequestChannel used for asynchronous request-reply communication within the same memory space. It allows a sender to publish a request of type R and receive a reply of type V via a callback.

Key Operations:

• Subscribing to requests: Use subscribe to listen for incoming requests. You can provide a Callback<Request<R, V>> to handle the request, and optionally a Callback<SessionClosed<R>> to be notified when a request session is closed (disposed).
• Publishing requests: Use publish to send a request. You must provide the DisposingExecutor (target) where the reply should be executed, the request object R, and a Callback<V> to handle the response.
• Lifecycle Management: Both subscribing and publishing return a Disposable. Disposing the returned object cleans up the subscription or cancels the request session.

// Example usage of MemoryRequestChannel
MemoryRequestChannel<String, Integer> channel = new MemoryRequestChannel<>();

// 1. Subscribe to requests
Disposable subscription = channel.subscribe(executor, new Callback<Request<String, Integer>>() {
    @Override
    public void onMessage(Request<String, Integer> request) {
        // Handle the request and reply
        request.reply(42);
    }
});

// 2. Publish a request
Disposable requestHandle = channel.publish(executor, "Hello", new Callback<Integer>() {
    @Override
    public void onMessage(Integer reply) {
        System.out.println("Received reply: " + reply);
    }
});

// 3. Cleanup
subscription.dispose();
requestHandle.dispose();

The SingleReply.publish method allows you to send a request through a RequestChannel and expect exactly one response. It wraps an AsyncRequest and automatically handles the extraction of the first message from the response list, passing it directly to your provided Callback.

There are two ways to use this:

1. Basic publish: Sends a request and triggers the reply callback when the first response arrives.
2. Publish with timeout: Sends a request and triggers the reply callback on response, or executes an onTimeout Runnable if the specified timeout period expires before a response is received.

// Basic usage: publish a request and handle the single response
SingleReply.publish(
    fiber, 
    channel, 
    request, 
    new Callback<V>() {
        @Override
        public void onMessage(V message) {
            // Handle the single response here
        }
    }
);

// Usage with timeout: handle response or execute onTimeout
SingleReply.publish(
    fiber, 
    channel, 
    request, 
    new Callback<V>() {
        @Override
        public void onMessage(V message) {
            // Handle the single response here
        }
    },
    5, TimeUnit.SECONDS, 
    () -> {
        // Handle timeout logic here
    }
);

The AsyncRequest class is used to send a request through a RequestChannel and manage how responses are collected. You can configure it to wait for a specific number of responses using setResponseCount(int responses) or to collect responses within a specific timeframe using setTimeout(Callback<List<V>> onTimeout, long time, TimeUnit unit).

To execute the request, call publish(RequestChannel<R, V> channel, R req, Callback<List<V>> onResponse). This returns a Disposable which should be used to manage the lifecycle of the request.

// Example: Requesting multiple responses with a timeout
AsyncRequest<MyRequest, MyResponse> request = new AsyncRequest<>(targetFiber)
    .setResponseCount(3)
    .setTimeout(responses -> {
        System.out.println("Timeout reached or batch complete: " + responses);
    }, 5, TimeUnit.SECONDS);

Disposable subscription = request.publish(myChannel, myRequest, responses -> {
    System.out.println("Received batch: " + responses);
});