advantageous / reakt Goto Github PK

Promise<Result> resultPromise = Promise.<Result>promise();

Promise<Employee> employeePromise = resultPromise.<Employee> thenMap((promise)-> {
         Result result = promise.get(); 
         Employee e = new Employee(result.getCol("id"), result.getCol("firstName"), ...);
         return e;
});

Promise<Employee> employeePromise = Promise.<Result>promise().  
                                  <Employee> thenMap((promise)-> {
                        Result r = promise.get(); 
                        return new Employee(r.getCol("id"), r.getCol("firstName"), ...);
});

    @Override
    public <U> Promise<U> thenMap(Function<? super T, ? extends U> mapper) {
        final Promise<U> mappedPromise = Promises.promise();
        this.whenComplete(p -> {
            final BlockingPromise<T> promise = (BlockingPromise) p;
            if (promise._success()) {
                final T t = promise.result.get().get();
                final U mapped = mapper.apply(t);
                mappedPromise.reply(mapped);
            } else {
                mappedPromise.reject(promise.cause());
            }
        });
        return mappedPromise;
    }

    @Override
    public <U> Promise<U> thenMap(Function<? super T, ? extends U> mapper) {
        final Promise<U> mappedPromise = Promises.blockingPromise();
        this.whenComplete(p -> {
            final BlockingPromise<T> promise = (BlockingPromise) p;
            if (promise._success()) {
                final T t = promise.result.get().get();
                final U mapped = mapper.apply(t);
                mappedPromise.reply(mapped);
            } else {
                mappedPromise.reject(promise.cause());
            }
        });
        return mappedPromise;
    }

        employeeService.lookupEmployee("123")
               .then((employee)-> {...}).catchError(...).invoke();

        employeeService.lookupEmployee("123")
               .reactor(mgmt.reactor())
               .then((employee)-> {...}).catchError(...).invoke();

        repository.fetchRawMetricsSince(snapshot.timestamp)
                .then { rawMetrics -> becomeMaster(snapshot, callback, rawMetrics) }
                .catchError { error -> rejectBeingMaster(snapshot, callback, error) }
                .invokeWithPromise(mgmt.reactor().promise());

breaker1.ifBrokeOrBroke(breaker2) {
     //do something if either breaker is broken
}

        reactor
                .all(Duration.ofSeconds(30),
                        saveToTodoTablePromise, saveToLookupTable)
                .catchError(returnPromise::reject)
                .then( v-> returnPromise.resolve(true))
                .invoke();

/* 
 * A {@code Callback} is a {@code Consumer} and can be used anywhere a consumer is used.
 * This is for easy integration with non-Reakt libs and code bases.
 * <p>
 *
 * @param <T> type of result returned from callback
 */
public interface Callback<T> extends Consumer<T> {
    ...

    /**
     * Bridge between Consumer world and Callback world
     * Performs this operation on the given argument.
     *
     * @param t the input argument
     */
    @Override
    default void accept(T t) {
        reply(t);
    }

    /**
     * Used to convert the error handling of the callback or promise
     * into a Consumer so you can easily integrate with non-Reakt code.
     *
     * @return Consumer version of error handling.
     */
    default Consumer<Throwable> errorConsumer() {
        return this::reject;
    }

    /**
     * Used to easily cast this callback to a consumer.
     *
     * @return Consumer version of this callback.
     */
    default Consumer<T> consumer() {
        return this;
    }
}

class Promises

    /**
     * Returns a String promise
     * Added to make static imports possible.
     *
     * @return returns a string promise
     */
    static Promise<String> promiseString() {
        return new BasePromise<>();
    }

    /**
     * Returns a Integer promise
     * Added to make static imports possible.
     *
     * @return returns an int promise
     */
    static Promise<Integer> promiseInt() {
        return new BasePromise<>();
    }


    /**
     * Returns a Long promise
     * Added to make static imports possible.
     *
     * @return returns an long promise
     */
    static Promise<Long> promiseLong() {
        return new BasePromise<>();
    }

    /**
     * Returns a Double promise
     * Added to make static imports possible.
     *
     * @return returns an double promise
     */
    static Promise<Double> promiseDouble() {
        return new BasePromise<>();
    }


    /**
     * Returns a Float promise
     * Added to make static imports possible.
     *
     * @return returns an float promise
     */
    static Promise<Float> promiseFloat() {
        return new BasePromise<>();
    }

    /**
     * Returns a void promise for notify of outcome but no value returned.
     * <p>
     * Callback replyDone can be used instead of replay on service side.
     *
     * @return void promise
     */
    static Promise<Void> promiseNotify() {
        return new BasePromise<>();
    }

    /**
     * Boolean promise
     * Added to make static imports possible.
     *
     * @return promises a boolean
     */
    static Promise<Boolean> promiseBoolean() {
        return new BasePromise<>();
    }


    /**
     * Generic promise.
     * Added to make static imports possible.
     *
     * @param cls type
     * @param <T> promise of a result of T
     * @return new Promise of type T
     */
    @SuppressWarnings("unused")
    static <T> Promise<T> promise(Class<T> cls) {
        return new BasePromise<>();
    }


    /**
     * Generic list promise.
     * Added to make static imports possible.
     *
     * @param componentType component type of list
     * @param <T>           promise a list of type T
     * @return new Promise for a list of type T
     */
    @SuppressWarnings("unused")
    static <T> Promise<List<T>> promiseList(Class<T> componentType) {
        return new BasePromise<>();
    }

    /**
     * Generic collection promise.
     * Added to make static imports possible.
     *
     * @param componentType component type of collection
     * @param <T>           promise a collection of type T
     * @return new Promise for a collection of type T
     */
    @SuppressWarnings("unused")
    static <T> Promise<Collection<T>> promiseCollection(Class<T> componentType) {
        return new BasePromise<>();
    }

    /**
     * Generic map promise.
     * Added to make static imports possible.
     *
     * @param keyType   type of map key
     * @param valueType type of map value
     * @param <K>       promise a map of  key type K
     * @param <V>       promise a map of  value type V
     * @return new Promise for a collection of type T
     */
    @SuppressWarnings("unused")
    static <K, V> Promise<Map<K, V>> promiseMap(Class<K> keyType, Class<V> valueType) {
        return new BasePromise<>();
    }

    /**
     * Generic set promise.
     * Added to make static imports possible.
     *
     * @param componentType component type of set
     * @param <T>           promise a set of type T
     * @return new Promise for a set of type T
     */
    @SuppressWarnings("unused")
    static <T> Promise<Set<T>> promiseSet(Class<T> componentType) {
        return new BasePromise<>();
    }

+    /**
 +     * Used for testing and legacy integration.
 +     * This turns an async promise into a blocking promise.
 +     * @return blocking promise
 +     */
 +    default Promise<T> invokeAsBlockingPromise() {
 +        Promise<T> blockingPromise = Promises.blockingPromise();
 +        this.invokeWithPromise(blockingPromise);
 +        return blockingPromise;
 +    }
 +
 +
 +    /**
 +     * Used for testing and legacy integration.
 +     * This turns an async promise into a blocking promise.
 +     * @param duration duration
 +     * @return blocking promise
 +     */
 +    default Promise<T> invokeAsBlockingPromise(Duration duration) {
 +        Promise<T> blockingPromise = Promises.blockingPromise(duration);
 +        this.invokeWithPromise(blockingPromise);
 +        return blockingPromise;
 +    }