InterceptorDemoAsFunctions.java

/*
 * Copyright 2015 Netflix, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package io.reactivex.netty.examples.tcp;

import rx.Observable;
import rx.functions.Func1;

public final class InterceptorDemoAsFunctions {

    private InterceptorDemoAsFunctions() {
    }

    public interface Server<I, O> extends Func1<I, Observable<O>> {
    }

    public static <O> Server<Integer, O> modifyRequest(Server<String, O> server) {
        return request -> server.call(request.toString());
    }

    public static <I, O> Server<I, String> modifyResponse(Server<I, O> server) {
        return i -> server.call(i).map(t1 -> "Hello World! " + t1);
    }

    public static <I, O> Server<I, O> log(Server<I, O> server) {
        return request -> {
            System.out.println("Received request for logging: " + request);
            return server.call(request);
        };
    }

    public static <I, O> Server<I, O> shortCircuit(Server<I, O> server, Func1<I, Boolean> predicate, O defaultValue) {
        return request -> {
            if (predicate.call(request)) {
                return Observable.just(defaultValue);
            }
            return server.call(request);
        };
    }

    public static void main(String[] args) {

        final Observable<Integer> socket = Observable.range(1, 10);

        Server<Integer, String> server =
                log(shortCircuit(modifyResponse(
                                         modifyRequest(request -> Observable.just(Integer.valueOf(request)))),
                                 request -> request % 2 == 0, "Hello Cached"));

        socket.flatMap(server::call).toBlocking().forEach(System.out::println);
    }
}