jetlang/core

Jetlang provides a high performance java threading library. The library is based upon Retlang.

The library is a complement to the java.util.concurrent package introduced in 1.5 and should be used for message based concurrency similar to event based actors in Scala.

The library does not provide remote messaging capabilities. It is designed specifically for high performance in-memory messaging. Features

• All messages to a particular Fiber are delivered sequentially. Components can easily keep state without synchronizing data access or worrying about thread races.
• Single Fiber interface that can be backed by a dedicated thread or a thread pool.
• Supports single or multiple subscribers for messages.
• Subscriptions for single events or event batching
• Single or recurring event scheduling
• High performance design optimized for low latency and high scalability
• Publishing is thread safe, allowing easy integration with other threading models.
• Low Lock Contention - Minimizing lock contention is critical for performance. Other concurrency solutions are limited by a single lock typically on a central thread pool or message queue. Jetlang is optimized for low lock contention. Without a central bottleneck, performance easily scales to the needs of the application.
• Powerful Async Request/Reply Support
• Single jar with no dependencies except the jdk (1.6+)
• Integrates with any JVM language - jruby, scala, clojure, groovy, etc
• Distributed Messaging - Jetlang Remoting

Example

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();

Channel Example

    // 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();

##Using with Maven

Jetlang is in the central maven repository as of 0.2.9

https://repo1.maven.org/maven2/org/jetlang/jetlang/

To use Jetlang, add the following dependency. Replace DESIRED_VERSION_HERE with the version you would like to use.

    <project>
        ...
        <dependencies>
            ...
            <dependency>
                <groupId>org.jetlang</groupId>
                <artifactId>jetlang</artifactId>
                <version>_DESIRED_VERSION_HERE_</version>
            </dependency>
            ...
        </dependencies>
        ...
    </project>

##Dev Mailing List

https://groups.google.com/forum/#!forum/jetlang-dev

Releasing with Maven

mvn release:prepare release:perform

core's Issues

ThreadFiber executes pending commands after dispose, PoolFiber does not

We've been using jetlang 0.2.5 for few years and decided to update all of the third party libraries used by our framework. After updating to version 0.2.12, some of our unit tests failed and I investigated the reasons.

The cause is a change from 2012 in PoolFiber. Since then, pending commands will not be executed anymore after calling dispose() on the fiber. This is in contrast to the ThreadFiber, that still executes pending commands. Is this an intentional difference in behaviour or just an unwanted side effect caused by a fix? Which kind of behavior should be expected from a fiber? Should all fibers behave the same way?

In our case, we have a callback that is called when the fiber is stopped. During and after this callback, no other commands should be executed. Until now, I cancelled all subscriptions and added the callback to the execution queue of the fiber, then disposed it. This still works with ThreadFiber, but not with PoolFiber (probably prevented by isRunning in https://github.com/jetlang/core/blob/master/src/main/java/org/jetlang/fibers/PoolFiber.java#L61). I fixed this by disposing the fiber only after the callback was invoked (using a CountDownLatch and blocking the calling thread), so for us, it is no big deal. I just wonder whether all fibers should behave the same way.

PoolFiber backed by same thread never fulfills tasks after call to `start`

Hi,

The following code will never terminate:

Fiber fiber = new PoolFiberFactory(Runnable::run).create();
fiber.start();

CompletableFuture<Integer> future = supplyAsync(() -> 1, fiber)
    .thenCompose(x -> supplyAsync(() -> x + 2, fiber));

System.out.println(future.join());

Although the backing executor on its own works as expected:

Executor sameThread = Runnable::run;

CompletableFuture<Integer> future = supplyAsync(() -> 1, sameThread)
    .thenCompose(x -> supplyAsync(() -> x + 2, sameThread));

System.out.println(future.join());

At least one use case for this is to avoid instantiating a bunch of unnecessary threads in test code.

The above behavior feels like a bug -- is there any reason it would be expected or desired behavior?

@FunctionalInterface on functional interfaces

Jetlang contains a few widely-used interfaces which contain a single abstract method with no method-level type parameters. These are therefore "functional interfaces", and can be implemented with lambda expressions and method references. The primary example would be Publisher, but Filter, Callback, and Converter are others. In my code, all of these are often implemented by lambdas and method references, probably even a majority of the time.

Java has an annotation, @FunctionalInterface, to mark functional interfaces. It's purely a marker annotation, and has no functional consequences, so these interfaces are just as usable without it as with it.

With one substantial exception: IntelliJ will only consider an interface as the target of an Extract Functional Parameter or Extract Functional Variable refactoring if it is annotated with @FunctionalInterface! I lean very heavily on IntelliJ's refactorings, including that one, and when i'm doing Jetlang plumbing, it's a definite drag not to be able to quickly extract something into a Publisher or whatever.

So, could we add @FunctionalInterface to functional interfaces where possible?

I'm happy to submit a PR to that if so.