package com.github.blackrush.acara;

import org.fungsi.Unit;
import org.fungsi.concurrent.Future;
import org.fungsi.concurrent.Futures;
import org.fungsi.concurrent.Worker;
import org.fungsi.function.UnsafeRunnable;
import org.fungsi.function.UnsafeSupplier;
import org.junit.Test;

import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.reflect.Method;
import java.util.stream.Stream;

import static org.junit.Assert.assertEquals;

public class JavaEventBus {

    static class TheEvent {
        final String name;
        TheEvent(String name) { this.name = name; }
    }

    static class FunkyEvent extends TheEvent {
        FunkyEvent(String name) { super(name); }
    }

    @Retention(RetentionPolicy.RUNTIME)
    static @interface Funky{}

    static class FunkyListenerBuilder extends JavaListenerBuilder {
        @Override
        protected Stream<Listener> scan(Object o, Method method) {
            if (!method.isAnnotationPresent(Funky.class)) {
                return Stream.empty();
            }
            JavaListener listener = new JavaListener(new JavaEventMetadata(FunkyEvent.class), o, method) {
                @Override
                protected Object invoke(Object state, Method behavior, Object event) throws Throwable {
                    FunkyEvent evt = (FunkyEvent) event;
                    return behavior.invoke(state, evt.name);
                }
            };
            return Stream.of(listener);
        }
    }

    static class TheListener {
        @Listen
        public String listen(TheEvent evt) {
            return String.format("Hello, %s!", evt.name);
        }

        @Funky
        public String funky(String name) {
            return String.format("Wassup %s?", name);
        }
    }

    @Test
    public void testSubscribePublishRevoke() throws Exception {

        EventBus eventBus = new EventBusImpl(
                new JavaEventMetadataBuilder(),
                new JavaListenerBuilder().concat(new FunkyListenerBuilder()),
                new SameThreadWorker());

        Subscription sub = eventBus.subscribe(new TheListener());
        String msg1 = (String) eventBus.publish(new TheEvent("World")).get().get(0);
        String msg2 = (String) eventBus.publish(new FunkyEvent("dawg")).get().get(0);
        sub.revoke();

        assertEquals("first message", "Hello, World!", msg1);
        assertEquals("second message", "Wassup dawg?", msg2);
    }

    class SameThreadWorker implements Worker {

        @Override
        public <T> Future<T> execute(UnsafeSupplier<Future<T>> fn) {
            try {
                return fn.get();
            } catch (Throwable throwable) {
                throw new RuntimeException(throwable);
            }
        }

        @Override
        public <T> Future<T> submit(UnsafeSupplier<T> fn) {
            try {
                return Futures.success(fn.get());
            } catch (Throwable throwable) {
                throw new RuntimeException(throwable);
            }
        }

        @Override
        public Future<Unit> cast(UnsafeRunnable fn) {
            try {
                fn.run();
                return Futures.unit();
            } catch (Throwable throwable) {
                throw new RuntimeException(throwable);
            }
        }
    }
}

@@ -41,221 +41,93 @@
 ort 
-org.fungsi.Unit;%0Aimport org.fungsi.concurrent.Future;%0Aimport org.fungsi.concurrent.Futures;%0Aimport org.fungsi.concurrent.Worker;%0Aimport org.fungsi.function.UnsafeRunnable;%0Aimport org.fungsi.function.UnsafeSupplier
+com.google.common.util.concurrent.MoreExecutors;%0Aimport org.fungsi.concurrent.Workers
 ;%0Aim
@@ -1929,31 +1929,53 @@
     
-new SameThreadWorke
+Workers.wrap(MoreExecutors.directExecuto
 r())
+)
 ;%0A%0A 
@@ -2368,870 +2368,6 @@
  %7D%0A%0A
-    class SameThreadWorker implements Worker %7B%0A%0A        @Override%0A        public %3CT%3E Future%3CT%3E execute(UnsafeSupplier%3CFuture%3CT%3E%3E fn) %7B%0A            try %7B%0A                return fn.get();%0A            %7D catch (Throwable throwable) %7B%0A                throw new RuntimeException(throwable);%0A            %7D%0A        %7D%0A%0A        @Override%0A        public %3CT%3E Future%3CT%3E submit(UnsafeSupplier%3CT%3E fn) %7B%0A            try %7B%0A                return Futures.success(fn.get());%0A            %7D catch (Throwable throwable) %7B%0A                throw new RuntimeException(throwable);%0A            %7D%0A        %7D%0A%0A        @Override%0A        public Future%3CUnit%3E cast(UnsafeRunnable fn) %7B%0A            try %7B%0A                fn.run();%0A                return Futures.unit();%0A            %7D catch (Throwable throwable) %7B%0A                throw new RuntimeException(throwable);%0A            %7D%0A        %7D%0A    %7D%0A
 %7D%0A

Remove useless SameThreadWorker inside test