fix compiler flow issues (#414)

replace lambdas with method references
replace statement lambda with expression lambda
redundant local variable
Unchecked assignment: 'org.reactivestreams.Subscriber' to 'org.reactivestreams.Subscriber<? super io.rsocket.Payload>'
Unchecked assignment: 'org.reactivestreams.Subscriber' to 'org.reactivestreams.Subscriber<? super java.lang.Void>'
The declared exception 'Exception' is never thrown

Author: qweek

rsocket-core/src/jmh/java/io/rsocket/RSocketPerf.java

@@ -69,7 +69,7 @@ public void requestStreamHello1000(Input input) {
   @Benchmark
   public void fireAndForgetHello(Input input) {
     // this is synchronous so we don't need to use a CountdownLatch to wait
-    input.client.fireAndForget(Input.HELLO_PAYLOAD).subscribe(input.blackHoleSubscriber);
+    input.client.fireAndForget(Input.HELLO_PAYLOAD).subscribe(input.voidSubscriber);
   }
 
   @State(Scope.Benchmark)
@@ -155,37 +155,42 @@ public Mono<Void> onClose() {
                   return Mono.just(closeable);
                 });
 
-    Subscriber blackHoleSubscriber;
+    Subscriber<Payload> blackHoleSubscriber;
+    Subscriber<Void> voidSubscriber;
 
     RSocket client;
 
     @Setup
     public void setup(Blackhole bh) {
-      blackHoleSubscriber =
-          new Subscriber() {
-            @Override
-            public void onSubscribe(Subscription s) {
-              s.request(Long.MAX_VALUE);
-            }
-
-            @Override
-            public void onNext(Object o) {
-              bh.consume(o);
-            }
-
-            @Override
-            public void onError(Throwable t) {
-              t.printStackTrace();
-            }
-
-            @Override
-            public void onComplete() {}
-          };
+      blackHoleSubscriber = subscriber(bh);
+      voidSubscriber = subscriber(bh);
 
       client =
           RSocketFactory.connect().transport(() -> Mono.just(clientConnection)).start().block();
 
       this.bh = bh;
     }
+
+    private <T> Subscriber<T> subscriber(Blackhole bh) {
+      return new Subscriber<T>() {
+        @Override
+        public void onSubscribe(Subscription s) {
+          s.request(Long.MAX_VALUE);
+        }
+
+        @Override
+        public void onNext(T o) {
+          bh.consume(o);
+        }
+
+        @Override
+        public void onError(Throwable t) {
+          t.printStackTrace();
+        }
+
+        @Override
+        public void onComplete() {}
+      };
+    }
   }
 }

rsocket-examples/src/test/java/io/rsocket/integration/TestingStreaming.java

@@ -15,18 +15,18 @@
 import reactor.core.publisher.Mono;
 
 public class TestingStreaming {
-  Supplier<ServerTransport<? extends Closeable>> serverSupplier =
+  private Supplier<ServerTransport<? extends Closeable>> serverSupplier =
       () -> LocalServerTransport.create("test");
 
-  Supplier<ClientTransport> clientSupplier = () -> LocalClientTransport.create("test");
+  private Supplier<ClientTransport> clientSupplier = () -> LocalClientTransport.create("test");
 
   @Test(expected = ApplicationException.class)
-  public void testRangeButThrowException() throws Exception {
+  public void testRangeButThrowException() {
     Closeable server = null;
     try {
       server =
           RSocketFactory.receive()
-              .errorConsumer(t -> t.printStackTrace())
+              .errorConsumer(Throwable::printStackTrace)
               .acceptor(
                   (connectionSetupPayload, rSocket) -> {
                     AbstractRSocket abstractRSocket =
@@ -65,12 +65,12 @@ public Flux<Payload> requestStream(Payload payload) {
   }
 
   @Test
-  public void testRangeOfConsumers() throws Exception {
+  public void testRangeOfConsumers() {
     Closeable server = null;
     try {
       server =
           RSocketFactory.receive()
-              .errorConsumer(t -> t.printStackTrace())
+              .errorConsumer(Throwable::printStackTrace)
               .acceptor(
                   (connectionSetupPayload, rSocket) -> {
                     AbstractRSocket abstractRSocket =
@@ -102,30 +102,21 @@ public Flux<Payload> requestStream(Payload payload) {
     }
   }
 
-  public Flux<Payload> consumer(String s) {
-    Mono<RSocket> test =
-        RSocketFactory.connect()
-            .errorConsumer(t -> t.printStackTrace())
-            .transport(clientSupplier)
-            .start();
-
-    Flux<Payload> payloadFlux =
-        test.flatMapMany(
+  private Flux<Payload> consumer(String s) {
+    return RSocketFactory.connect()
+        .errorConsumer(Throwable::printStackTrace)
+        .transport(clientSupplier)
+        .start()
+        .flatMapMany(
             rSocket -> {
               AtomicInteger count = new AtomicInteger();
               return Flux.range(1, 100)
-                  .flatMap(
-                      i -> {
-                        return rSocket.requestStream(new PayloadImpl("i -> " + i)).take(100);
-                      },
-                      1);
+                  .flatMap(i -> rSocket.requestStream(new PayloadImpl("i -> " + i)).take(100), 1);
             });
-
-    return payloadFlux;
   }
 
   @Test
-  public void testSingleConsumer() throws Exception {
+  public void testSingleConsumer() {
     Closeable server = null;
 
     try {