Publishers cleanup and new functions

reactivesocket-core/src/main/java/io/reactivesocket/RequestHandler.java

@@ -13,6 +13,7 @@
 package io.reactivesocket;
 
 import io.reactivesocket.internal.PublisherUtils;
+import io.reactivesocket.internal.Publishers;
 import org.reactivestreams.Publisher;
 
 import java.util.function.Function;
@@ -28,13 +29,13 @@ public abstract class RequestHandler {
         payload -> PublisherUtils.errorPayload(new RuntimeException("No 'requestSubscription' handler"));
 
     private static final Function<Payload, Publisher<Void>> NO_FIRE_AND_FORGET_HANDLER =
-        payload -> PublisherUtils.errorVoid(new RuntimeException("No 'fireAndForget' handler"));
+        payload -> Publishers.error(new RuntimeException("No 'fireAndForget' handler"));
 
     private static final Function<Publisher<Payload>, Publisher<Payload>> NO_REQUEST_CHANNEL_HANDLER =
         payloads -> PublisherUtils.errorPayload(new RuntimeException("No 'requestChannel' handler"));
 
     private static final Function<Payload, Publisher<Void>> NO_METADATA_PUSH_HANDLER =
-        payload -> PublisherUtils.errorVoid(new RuntimeException("No 'metadataPush' handler"));
+        payload -> Publishers.error(new RuntimeException("No 'metadataPush' handler"));
 
     public abstract Publisher<Payload> handleRequestResponse(final Payload payload);
 

reactivesocket-core/src/main/java/io/reactivesocket/internal/CancellableSubscriber.java

@@ -0,0 +1,61 @@
+/*
+ * Copyright 2016 Netflix, Inc.
+ * <p>
+ *  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
+ *  <p>
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *  <p>
+ *  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.reactivesocket.internal;
+
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
+
+abstract class CancellableSubscriber<T> implements Subscriber<T> {
+
+    private Subscription s;
+    private boolean cancelled;
+
+    @Override
+    public void onSubscribe(Subscription s) {
+        boolean _cancel = false;
+        synchronized (this) {
+            this.s = s;
+            if (cancelled) {
+                _cancel = true;
+            }
+        }
+
+        if (_cancel) {
+            _unsafeCancel();
+        }
+    }
+
+    public void cancel() {
+        boolean _cancel = false;
+        synchronized (this) {
+            cancelled = true;
+            if (s != null) {
+                _cancel = true;
+            }
+        }
+
+        if (_cancel) {
+            _unsafeCancel();
+        }
+    }
+
+    protected void doAfterCancel() {
+        // NoOp by default.
+    }
+
+    private void _unsafeCancel() {
+        s.cancel();
+        doAfterCancel();
+    }
+}

reactivesocket-core/src/main/java/io/reactivesocket/internal/PublisherUtils.java

@@ -105,69 +105,6 @@ public void cancel() {
         };
     }
 
-    public static final Publisher<Void> errorVoid(Throwable e) {
-        return (Subscriber<? super Void> s) -> {
-            s.onSubscribe(new Subscription() {
-
-                @Override
-                public void request(long n) {
-                }
-
-                @Override
-                public void cancel() {
-                    // ignoring as nothing to do
-                }
-
-            });
-            s.onError(e);
-
-        };
-    }
-
-    public static final Publisher<Frame> just(Frame frame) {
-        return (Subscriber<? super Frame> s) -> {
-            s.onSubscribe(new Subscription() {
-
-                boolean completed = false;
-
-                @Override
-                public void request(long n) {
-                    if (!completed && n > 0) {
-                        completed = true;
-                        s.onNext(frame);
-                        s.onComplete();
-                    }
-                }
-
-                @Override
-                public void cancel() {
-                    // ignoring as nothing to do
-                }
-
-            });
-
-        };
-    }
-
-    public static final <T> Publisher<T> empty() {
-        return (Subscriber<? super T> s) -> {
-            s.onSubscribe(new Subscription() {
-
-                @Override
-                public void request(long n) {
-                }
-
-                @Override
-                public void cancel() {
-                    // ignoring as nothing to do
-                }
-
-            });
-            s.onComplete(); // TODO confirm this is okay with ReactiveStream spec to send immediately after onSubscribe (I think so since no data is being sent so requestN doesn't matter)
-        };
-
-    }
-
     public static final Publisher<Frame> keepaliveTicker(final int interval, final TimeUnit timeUnit) {
         return (Subscriber<? super Frame> s) -> {
             s.onSubscribe(new Subscription()