InterceptorDemoAsInterfaces.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 InterceptorDemoAsInterfaces {

    private InterceptorDemoAsInterfaces() {
    }

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

    public interface Interceptor<I, O, II, OO> extends Func1<Server<I, O>, Server<II, OO>> { }

    public static class ServerLauncher<I, O> {/*Horrible name*/

        private final Server<I, O> server;

        public ServerLauncher(Server<I, O> server) {
            this.server = server;
        }

        public <II, OO> ServerLauncher<II, OO> intercept(Interceptor<I, O, II, OO> interceptor) {
            return new ServerLauncher<>(interceptor.call(server));
        }

        public static <I, O> ServerLauncher<I, O> newLauncher(Server<I, O> server) {
            return new ServerLauncher<>(server);
        }
    }

    public static void main(String[] args) {

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

        Server<Integer, String> theServer =
                ServerLauncher.<String, Integer>newLauncher(stringReq -> Observable.just(Integer.valueOf(stringReq)))
                              .intercept(server -> new Server<Integer, Integer>() { /*Request type converter*/
                                  @Override
                                  public Observable<Integer> call(Integer aInt) {
                                      return server.call(aInt.toString());
                                  }
                              })
                              .intercept(server -> new Server<Integer, String>() { /*Response type converter*/
                                  @Override
                                  public Observable<String> call(Integer aInt) {
                                      return server.call(aInt).map(respInt -> "Hello World! " + respInt.toString());
                                  }
                              })
                              .intercept(server -> new Server<Integer, String>() { /*Short-circuit*/
                                  @Override
                                  public Observable<String> call(Integer request) {
                                      if (request % 2 == 0) {
                                          return Observable.just("Hello from cache.");
                                      } else {
                                          return server.call(request);
                                      }
                                  }
                              })
                              .intercept(server -> new Server<Integer, String>() { /*Logging*/
                                  @Override
                                  public Observable<String> call(Integer request) {
                                      System.out.println("Received request for logging: " + request);
                                      return server.call(request);
                                  }
                              })
                        .server;

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

    }
}