### Create and Use ThreadFiber in Java

Source: https://context7.com/jetlang/core/llms.txt

Demonstrates how to create a ThreadFiber, which uses a dedicated thread for sequential message delivery. Tasks are executed on this thread, and the Fiber must be started before use and disposed afterward. This is useful for maintaining state without explicit synchronization.

```java
import org.jetlang.fibers.ThreadFiber;
import org.jetlang.fibers.Fiber;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

// Create a dedicated thread-backed Fiber
Fiber fiber = new ThreadFiber();
fiber.start();

final CountDownLatch latch = new CountDownLatch(2);

// Execute runnables on the Fiber's dedicated thread
Runnable task = new Runnable() {
    public void run() {
        System.out.println("Executing on: " + Thread.currentThread().getName());
        latch.countDown();
    }
};

// Queue tasks for sequential execution
fiber.execute(task);
fiber.execute(task);

// Wait for completion
latch.await(10, TimeUnit.SECONDS);

// Clean up resources
fiber.dispose();
```

--------------------------------

### Java: Resource Management with Disposable Interface

Source: https://context7.com/jetlang/core/llms.txt

Demonstrates how to use the Disposable interface in Java for managing resources like subscriptions and scheduled tasks within Jetlang. It shows how to start a Fiber, create a channel, subscribe to it, schedule a task, and then dispose of these resources to prevent leaks and ensure proper cleanup.

```java
import org.jetlang.channels.Channel;
import org.jetlang.channels.MemoryChannel;
import org.jetlang.core.Callback;
import org.jetlang.core.Disposable;
import org.jetlang.fibers.Fiber;
import org.jetlang.fibers.ThreadFiber;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.CountDownLatch;

Fiber fiber = new ThreadFiber();
fiber.start();

Channel<String> channel = new MemoryChannel<String>();

// Track active resources
final CountDownLatch subscriptionActive = new CountDownLatch(1);

Callback<String> callback = new Callback<String>() {
    public void onMessage(String msg) {
        System.out.println("Received: " + msg);
        subscriptionActive.countDown();
    }
};

// Subscription returns Disposable
Disposable subscription = channel.subscribe(fiber, callback);

// Scheduled task returns Disposable
Disposable scheduledTask = fiber.scheduleAtFixedRate(
    new Runnable() {
        public void run() {
            System.out.println("Periodic task running");
        }
    },
    0, 100, TimeUnit.MILLISECONDS
);

channel.publish("Test message");
subscriptionActive.await(5, TimeUnit.SECONDS);

// Cancel subscription - no more messages received
subscription.dispose();
channel.publish("This won't be received");

// Cancel scheduled task
scheduledTask.dispose();

// Fiber itself is Disposable - cleans up all resources
fiber.dispose();
```

--------------------------------

### Execute Runnable Tasks using ThreadFiber

Source: https://github.com/jetlang/core/blob/master/README.md

Demonstrates how to initialize a ThreadFiber, enqueue Runnable tasks for sequential execution, and properly dispose of the fiber to release resources.

```java
Fiber fiber = new ThreadFiber();
fiber.start();
final CountDownLatch latch = new CountDownLatch(2);
Runnable toRun = new Runnable(){
    public void run(){
        latch.countDown();
    }
};
//enqueue runnable for execution
fiber.execute(toRun);
//repeat to trigger latch a 2nd time
fiber.execute(toRun);
latch.await(10, TimeUnit.SECONDS);
//shutdown thread
fiber.dispose();
```

--------------------------------

### Create and Use MemoryChannel in Java

Source: https://context7.com/jetlang/core/llms.txt

Shows how to use MemoryChannel for publish/subscribe messaging in Jetlang. It demonstrates creating a typed channel, subscribing a Fiber to it with a callback, publishing messages, and unsubscribing. Publishing is thread-safe, and messages are delivered without serialization.

```java
import org.jetlang.channels.Channel;
import org.jetlang.channels.MemoryChannel;
import org.jetlang.core.Callback;
import org.jetlang.core.Disposable;
import org.jetlang.fibers.Fiber;
import org.jetlang.fibers.ThreadFiber;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

Fiber receiver = new ThreadFiber();
receiver.start();

final CountDownLatch latch = new CountDownLatch(1);

// Create a typed channel for String messages
Channel<String> channel = new MemoryChannel<String>();

// Define message handler
Callback<String> onMessage = new Callback<String>() {
    public void onMessage(String message) {
        System.out.println("Received: " + message);
        latch.countDown();
    }
};

// Subscribe and get a handle for unsubscribing
Disposable subscription = channel.subscribe(receiver, onMessage);

// Publish message - thread-safe from any thread
channel.publish("Hello, Jetlang!");

latch.await(10, TimeUnit.SECONDS);

// Unsubscribe when done
subscription.dispose();

// Further publishes won't reach disposed subscription
channel.publish("This won't be received");

receiver.dispose();
```

--------------------------------

### Broadcast Messages using CompositeChannel in Java

Source: https://context7.com/jetlang/core/llms.txt

Shows how to aggregate multiple channels into a single CompositeChannel. This allows for unified subscription to multiple event streams and broadcasting messages to all underlying channels simultaneously.

```java
import org.jetlang.channels.Channel;
import org.jetlang.channels.MemoryChannel;
import org.jetlang.channels.CompositeChannel;
import org.jetlang.core.Callback;
import org.jetlang.fibers.Fiber;
import org.jetlang.fibers.ThreadFiber;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

Fiber fiber = new ThreadFiber();
fiber.start();

Channel<String> channel1 = new MemoryChannel<String>();
Channel<String> channel2 = new MemoryChannel<String>();

Channel<String> composite = new CompositeChannel<String>(channel1, channel2);

final CountDownLatch latch = new CountDownLatch(2);

Callback<String> handler = new Callback<String>() {
    public void onMessage(String message) {
        System.out.println("Received: " + message);
        latch.countDown();
    }
};

composite.subscribe(fiber, handler);
composite.publish("Broadcast message");

latch.await(5, TimeUnit.SECONDS);
fiber.dispose();
```

--------------------------------

### Schedule tasks with Jetlang Fibers

Source: https://context7.com/jetlang/core/llms.txt

Demonstrates how to perform single delayed executions and recurring tasks using the Fiber scheduling API. All scheduled tasks return a Disposable object for lifecycle management and cancellation.

```java
import org.jetlang.fibers.Fiber;
import org.jetlang.fibers.ThreadFiber;
import org.jetlang.core.Disposable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

Fiber fiber = new ThreadFiber();
fiber.start();

// Single delayed execution
final CountDownLatch singleLatch = new CountDownLatch(1);
Runnable delayedTask = new Runnable() {
    public void run() {
        System.out.println("Executed after 100ms delay");
        singleLatch.countDown();
    }
};
Disposable singleSchedule = fiber.schedule(delayedTask, 100, TimeUnit.MILLISECONDS);
singleLatch.await(5, TimeUnit.SECONDS);

// Recurring execution at fixed rate
final AtomicInteger counter = new AtomicInteger(0);
final CountDownLatch recurringLatch = new CountDownLatch(5);
Runnable recurringTask = new Runnable() {
    public void run() {
        int count = counter.incrementAndGet();
        System.out.println("Recurring execution #" + count);
        recurringLatch.countDown();
    }
};

// Execute every 50ms after initial 10ms delay
Disposable recurringSchedule = fiber.scheduleAtFixedRate(
    recurringTask,
    10,    // initial delay
    50,    // period
    TimeUnit.MILLISECONDS
);

recurringLatch.await(5, TimeUnit.SECONDS);

// Cancel recurring task
recurringSchedule.dispose();

fiber.dispose();
```

--------------------------------

### Implement Inter-thread Messaging with Channels

Source: https://github.com/jetlang/core/blob/master/README.md

Shows how to create a MemoryChannel to pass messages between threads. It uses a Fiber to process subscriptions and a Callback to handle incoming messages asynchronously.

```java
Fiber receiver = new ThreadFiber();
receiver.start();
final CountDownLatch latch = new CountDownLatch(1);
Channel<String> channel = new MemoryChannel<String>();
Callback<String> onMsg = new Callback<String>() {
    public void onMessage(String message) {
        latch.countDown();
    }
};
channel.subscribe(receiver, onMsg);
channel.publish("Hello");
latch.await(10, TimeUnit.SECONDS);
receiver.dispose();
```

--------------------------------

### Create and Use PoolFiberFactory in Java

Source: https://context7.com/jetlang/core/llms.txt

Illustrates using PoolFiberFactory to create lightweight Fibers that share threads from an ExecutorService. This conserves thread resources while still ensuring sequential message delivery per Fiber. It shows creating multiple Fibers, subscribing them to a channel, and publishing messages.

```java
import org.jetlang.fibers.Fiber;
import org.jetlang.fibers.PoolFiberFactory;
import org.jetlang.channels.Channel;
import org.jetlang.channels.MemoryChannel;
import org.jetlang.core.Callback;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

// Create a shared thread pool and Fiber factory
ExecutorService threadPool = Executors.newCachedThreadPool();
PoolFiberFactory factory = new PoolFiberFactory(threadPool);

// Create multiple lightweight Fibers from the pool
Fiber fiber1 = factory.create();
Fiber fiber2 = factory.create();
fiber1.start();
fiber2.start();

Channel<String> channel = new MemoryChannel<String>();
final CountDownLatch latch = new CountDownLatch(2);

Callback<String> callback = new Callback<String>() {
    public void onMessage(String msg) {
        System.out.println("Received: " + msg + " on " + Thread.currentThread().getName());
        latch.countDown();
    }
};

// Both Fibers subscribe to same channel
channel.subscribe(fiber1, callback);
channel.subscribe(fiber2, callback);

// Publish message - both subscribers receive it
channel.publish("Hello from pool!");

latch.await(5, TimeUnit.SECONDS);

// Clean up
fiber1.dispose();
fiber2.dispose();
factory.dispose();
threadPool.shutdown();
```

--------------------------------

### Configure Maven Dependency for Jetlang

Source: https://github.com/jetlang/core/blob/master/README.md

Provides the XML configuration required to include the Jetlang library in a Maven project's dependencies.

```xml
<dependency>
    <groupId>org.jetlang</groupId>
    <artifactId>jetlang</artifactId>
    <version>_DESIRED_VERSION_HERE_</version>
</dependency>
```

--------------------------------

### Implement AsyncRequest for Multi-Response and Timeout Handling in Java

Source: https://context7.com/jetlang/core/llms.txt

Demonstrates using AsyncRequest to collect multiple responses from a single request and implementing timeout logic for unreliable responders. It utilizes Fibers for thread management and callbacks to handle incoming data or timeout events.

```java
import org.jetlang.channels.AsyncRequest;
import org.jetlang.channels.MemoryRequestChannel;
import org.jetlang.channels.RequestChannel;
import org.jetlang.channels.Request;
import org.jetlang.core.Callback;
import org.jetlang.fibers.Fiber;
import org.jetlang.fibers.ThreadFiber;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

Fiber requester = new ThreadFiber();
Fiber responder = new ThreadFiber();
requester.start();
responder.start();

RequestChannel<String, Integer> channel = new MemoryRequestChannel<String, Integer>();
final CountDownLatch done = new CountDownLatch(1);

Callback<Request<String, Integer>> onRequest = new Callback<Request<String, Integer>>() {
    public void onMessage(Request<String, Integer> request) {
        for (int i = 1; i <= 5; i++) {
            request.reply(i);
        }
    }
};
channel.subscribe(responder, onRequest);

Callback<List<Integer>> onReplies = new Callback<List<Integer>>() {
    public void onMessage(List<Integer> replies) {
        System.out.println("Received " + replies.size() + " replies: " + replies);
        done.countDown();
    }
};

AsyncRequest<String, Integer> asyncReq = new AsyncRequest<String, Integer>(requester);
asyncReq.setResponseCount(5);
asyncReq.publish(channel, "request", onReplies);

done.await(10, TimeUnit.SECONDS);

// Timeout example
RequestChannel<String, Integer> unreliableChannel = new MemoryRequestChannel<String, Integer>();
final CountDownLatch timeoutLatch = new CountDownLatch(1);

Callback<List<Integer>> onTimeout = new Callback<List<Integer>>() {
    public void onMessage(List<Integer> partialResults) {
        System.out.println("Timeout! Received " + partialResults.size() + " of expected responses");
        timeoutLatch.countDown();
    }
};

AsyncRequest<String, Integer> timedRequest = new AsyncRequest<String, Integer>(requester);
timedRequest.setTimeout(onTimeout, 100, TimeUnit.MILLISECONDS);
timedRequest.publish(unreliableChannel, "request", new Callback<List<Integer>>() {
    public void onMessage(List<Integer> responses) {}
});

timeoutLatch.await(5, TimeUnit.SECONDS);
requester.dispose();
responder.dispose();
```

--------------------------------

### KeyedBatchSubscriber: Batch Events by Key, Keep Latest (Java)

Source: https://context7.com/jetlang/core/llms.txt

KeyedBatchSubscriber batches messages based on a key, retaining only the most recent value for each key. This is useful for scenarios like stock updates where only the latest state matters. It requires a Fiber, a Callback for Map<K, V>, a time interval, TimeUnit, and a Converter to extract keys.

```java
import org.jetlang.channels.Channel;
import org.jetlang.channels.MemoryChannel;
import org.jetlang.channels.KeyedBatchSubscriber;
import org.jetlang.channels.Converter;
import org.jetlang.core.Callback;
import org.jetlang.fibers.Fiber;
import org.jetlang.fibers.ThreadFiber;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

Fiber fiber = new ThreadFiber();
fiber.start();

Channel<String> stockUpdates = new MemoryChannel<String>();
final CountDownLatch latch = new CountDownLatch(1);

// Callback receives map of latest values by key
Callback<Map<String, String>> batchCallback = new Callback<Map<String, String>>() {
    public void onMessage(Map<String, String> batch) {
        System.out.println("Latest prices: " + batch);
        if (batch.containsKey("GOOG")) {
            latch.countDown();
        }
    }
};

// Key resolver extracts stock symbol from "SYMBOL:PRICE" format
Converter<String, String> keyResolver = new Converter<String, String>() {
    public String convert(String msg) {
        return msg.split(":")[0];  // Return symbol as key
    }
};

KeyedBatchSubscriber<String, String> keyedBatch = new KeyedBatchSubscriber<String, String>(
    fiber,
    batchCallback,
    100,                   // flush interval
    TimeUnit.MILLISECONDS,
    keyResolver
);

stockUpdates.subscribe(keyedBatch);

// Multiple updates for same keys - only latest kept
stockUpdates.publish("AAPL:150.00");
stockUpdates.publish("GOOG:2800.00");
stockUpdates.publish("AAPL:151.00");  // Overwrites previous AAPL
stockUpdates.publish("AAPL:152.00");  // Overwrites again
stockUpdates.publish("GOOG:2810.00"); // Overwrites previous GOOG

// Batch will contain: {AAPL=AAPL:152.00, GOOG=GOOG:2810.00}
latch.await(5, TimeUnit.SECONDS);
fiber.dispose();
```

--------------------------------

### MemoryRequestChannel - Request/Reply Pattern (Java)

Source: https://context7.com/jetlang/core/llms.txt

Illustrates the MemoryRequestChannel for implementing an asynchronous request/reply pattern within a single JVM. It allows a requester to send a message and receive one or more replies on its Fiber. Dependencies include Jetlang channels, requests, callbacks, and fibers. It handles String requests and Integer replies.

```java
import org.jetlang.channels.MemoryRequestChannel;
import org.jetlang.channels.RequestChannel;
import org.jetlang.channels.Request;
import org.jetlang.core.Callback;
import org.jetlang.fibers.Fiber;
import org.jetlang.fibers.ThreadFiber;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

// Create two Fibers: one for requester, one for responder
Fiber requesterFiber = new ThreadFiber();
Fiber responderFiber = new ThreadFiber();
requesterFiber.start();
responderFiber.start();

// Create request channel: String requests, Integer replies
RequestChannel<String, Integer> channel = new MemoryRequestChannel<String, Integer>();

final CountDownLatch done = new CountDownLatch(1);

// Set up responder to handle requests
Callback<Request<String, Integer>> requestHandler = new Callback<Request<String, Integer>>() {
    public void onMessage(Request<String, Integer> request) {
        String req = request.getRequest();
        System.out.println("Received request: " + req);

        // Calculate response and send reply
        int response = req.length();
        request.reply(response);
    }
};
channel.subscribe(responderFiber, requestHandler);

// Set up reply handler
Callback<Integer> replyHandler = new Callback<Integer>() {
    public void onMessage(Integer reply) {
        System.out.println("Received reply: " + reply);
        done.countDown();
    }
};

// Publish request and receive reply
channel.publish(requesterFiber, "Hello, World!", replyHandler);

done.await(10, TimeUnit.SECONDS);

requesterFiber.dispose();
responderFiber.dispose();

```

--------------------------------

### BatchSubscriber: Collect and Deliver Events in Batches (Java)

Source: https://context7.com/jetlang/core/llms.txt

BatchSubscriber collects messages over a specified time interval and delivers them as a batch to a callback. This reduces processing overhead for high-frequency events. It requires a Fiber, a Callback for List<T>, a time interval, and a TimeUnit.

```java
import org.jetlang.channels.Channel;
import org.jetlang.channels.MemoryChannel;
import org.jetlang.channels.BatchSubscriber;
import org.jetlang.core.Callback;
import org.jetlang.fibers.Fiber;
import org.jetlang.fibers.ThreadFiber;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

Fiber fiber = new ThreadFiber();
fiber.start();

Channel<Integer> channel = new MemoryChannel<Integer>();
final CountDownLatch latch = new CountDownLatch(1);

// Callback receives batches of messages
Callback<List<Integer>> batchCallback = new Callback<List<Integer>>() {
    private int totalReceived = 0;

    public void onMessage(List<Integer> batch) {
        totalReceived += batch.size();
        System.out.println("Received batch of " + batch.size() +
            " messages: " + batch + " (total: " + totalReceived + ")");
        if (totalReceived >= 10) {
            latch.countDown();
        }
    }
};

// Create batch subscriber with 50ms flush interval
BatchSubscriber<Integer> batchSub = new BatchSubscriber<Integer>(
    fiber,
    batchCallback,
    50,                    // flush interval
    TimeUnit.MILLISECONDS
);

channel.subscribe(batchSub);

// Rapidly publish 10 messages - they will be batched
for (int i = 0; i < 10; i++) {
    channel.publish(i);
}

latch.await(5, TimeUnit.SECONDS);
fiber.dispose();
```

--------------------------------

### Filter messages in Jetlang Channels

Source: https://context7.com/jetlang/core/llms.txt

Shows how to implement the Filter interface to selectively process messages on a channel. Filtering occurs on the producer thread to optimize performance by preventing unnecessary context switches.

```java
import org.jetlang.channels.Channel;
import org.jetlang.channels.MemoryChannel;
import org.jetlang.channels.ChannelSubscription;
import org.jetlang.core.Callback;
import org.jetlang.core.Filter;
import org.jetlang.fibers.Fiber;
import org.jetlang.fibers.ThreadFiber;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

Fiber fiber = new ThreadFiber();
fiber.start();

Channel<Integer> channel = new MemoryChannel<Integer>();
final CountDownLatch latch = new CountDownLatch(2);

// Callback that only expects even numbers
Callback<Integer> onEvenNumber = new Callback<Integer>() {
    public void onMessage(Integer number) {
        System.out.println("Received even number: " + number);
        latch.countDown();
    }
};

// Filter that only passes even numbers
Filter<Integer> evenFilter = new Filter<Integer>() {
    public boolean passes(Integer msg) {
        return msg % 2 == 0;
    }
};

// Create filtered subscription
ChannelSubscription<Integer> filteredSub = new ChannelSubscription<Integer>(
    fiber, onEvenNumber, evenFilter
);
channel.subscribe(filteredSub);

// Publish mix of odd and even - only even received
channel.publish(1);  // Filtered out
channel.publish(2);  // Received
channel.publish(3);  // Filtered out
channel.publish(4);  // Received
channel.publish(5);  // Filtered out

latch.await(5, TimeUnit.SECONDS);
fiber.dispose();
```

--------------------------------

### LastSubscriber - Latest Value Only (Java)

Source: https://context7.com/jetlang/core/llms.txt

Demonstrates the LastSubscriber which ensures only the most recent message is processed within a given time interval. It's useful for scenarios like UI updates where stale data is undesirable. Dependencies include Jetlang channels and fibers. It takes a callback and a time unit for the flush interval.

```java
import org.jetlang.channels.Channel;
import org.jetlang.channels.MemoryChannel;
import org.jetlang.channels.LastSubscriber;
import org.jetlang.core.Callback;
import org.jetlang.fibers.Fiber;
import org.jetlang.fibers.ThreadFiber;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

Fiber fiber = new ThreadFiber();
fiber.start();

Channel<Integer> channel = new MemoryChannel<Integer>();
final AtomicInteger lastReceived = new AtomicInteger();
final CountDownLatch latch = new CountDownLatch(1);

// Callback receives only the latest value
Callback<Integer> onLatest = new Callback<Integer>() {
    public void onMessage(Integer value) {
        System.out.println("Received latest: " + value);
        lastReceived.set(value);
        if (value == 100) {
            latch.countDown();
        }
    }
};

// Only process latest value every 50ms
LastSubscriber<Integer> lastSub = new LastSubscriber<Integer>(
    fiber,
    onLatest,
    50,
    TimeUnit.MILLISECONDS
);

channel.subscribe(lastSub);

// Rapidly publish 100 values - most will be dropped
for (int i = 1; i <= 100; i++) {
    channel.publish(i);
}

latch.await(5, TimeUnit.SECONDS);
System.out.println("Final received value: " + lastReceived.get());  // Should be 100
fiber.dispose();

```

--------------------------------

### Maven Dependency for Jetlang

Source: https://context7.com/jetlang/core/llms.txt

Specifies the Maven dependency required to include the Jetlang library in a Java project. This configuration should be added to the pom.xml file to manage the library's version and availability.

```xml
<project>
    <dependencies>
        <dependency>
            <groupId>org.jetlang</groupId>
            <artifactId>jetlang</artifactId>
            <version>0.2.24</version>
        </dependency>
    </dependencies>
</project>
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.