Move Other Rx Classes to .internal.rx

Author: benjchristensen

src/main/java/io/reactivesocket/ReactiveSocket.java

@@ -26,10 +26,10 @@
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
 
-import io.reactivesocket.internal.CompositeCompletable;
-import io.reactivesocket.internal.CompositeDisposable;
 import io.reactivesocket.internal.Requester;
 import io.reactivesocket.internal.Responder;
+import io.reactivesocket.internal.rx.CompositeCompletable;
+import io.reactivesocket.internal.rx.CompositeDisposable;
 import io.reactivesocket.observable.Disposable;
 import io.reactivesocket.observable.Observable;
 import io.reactivesocket.observable.Observer;

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

@@ -27,6 +27,10 @@
 
 import io.reactivesocket.Frame;
 import io.reactivesocket.Payload;
+import io.reactivesocket.internal.rx.BackpressureHelper;
+import io.reactivesocket.internal.rx.BackpressureUtils;
+import io.reactivesocket.internal.rx.EmptySubscription;
+import io.reactivesocket.internal.rx.SubscriptionHelper;
 
 public class PublisherUtils {
 

src/main/java/io/reactivesocket/internal/Requester.java

@@ -38,6 +38,9 @@
 import io.reactivesocket.exceptions.CancelException;
 import io.reactivesocket.exceptions.Exceptions;
 import io.reactivesocket.exceptions.Retryable;
+import io.reactivesocket.internal.rx.BackpressureUtils;
+import io.reactivesocket.internal.rx.EmptyDisposable;
+import io.reactivesocket.internal.rx.EmptySubscription;
 import io.reactivesocket.observable.Disposable;
 import io.reactivesocket.observable.Observer;
 import uk.co.real_logic.agrona.collections.Int2ObjectHashMap;

src/main/java/io/reactivesocket/internal/Responder.java

@@ -28,6 +28,8 @@
 import org.reactivestreams.Subscription;
 
 import io.reactivesocket.exceptions.SetupException;
+import io.reactivesocket.internal.rx.EmptyDisposable;
+import io.reactivesocket.internal.rx.EmptySubscription;
 import io.reactivesocket.observable.Disposable;
 import io.reactivesocket.observable.Observer;
 import uk.co.real_logic.agrona.collections.Int2ObjectHashMap;

src/main/java/io/reactivesocket/internal/BackpressureHelper.java renamed to src/main/java/io/reactivesocket/internal/rx/BackpressureHelper.java

@@ -10,7 +10,7 @@
  * 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;
+package io.reactivesocket.internal.rx;
 
 import java.util.concurrent.atomic.*;
 

src/main/java/io/reactivesocket/internal/BackpressureUtils.java renamed to src/main/java/io/reactivesocket/internal/rx/BackpressureUtils.java

@@ -1,4 +1,4 @@
-package io.reactivesocket.internal;
+package io.reactivesocket.internal.rx;
 
 /**
  * Copyright 2015 Netflix, Inc.

src/main/java/io/reactivesocket/internal/BooleanDisposable.java renamed to src/main/java/io/reactivesocket/internal/rx/BooleanDisposable.java

@@ -1,4 +1,4 @@
-package io.reactivesocket.internal;
+package io.reactivesocket.internal.rx;
 
 import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
 

src/main/java/io/reactivesocket/internal/CompositeCompletable.java renamed to src/main/java/io/reactivesocket/internal/rx/CompositeCompletable.java

@@ -1,4 +1,4 @@
-package io.reactivesocket.internal;
+package io.reactivesocket.internal.rx;
 
 import java.util.HashSet;
 import java.util.Set;

src/main/java/io/reactivesocket/internal/CompositeDisposable.java renamed to src/main/java/io/reactivesocket/internal/rx/CompositeDisposable.java

@@ -1,4 +1,4 @@
-package io.reactivesocket.internal;
+package io.reactivesocket.internal.rx;
 
 import java.util.HashSet;
 import java.util.Set;

src/main/java/io/reactivesocket/internal/EmptyDisposable.java renamed to src/main/java/io/reactivesocket/internal/rx/EmptyDisposable.java

@@ -13,7 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package io.reactivesocket.internal;
+package io.reactivesocket.internal.rx;
 
 import io.reactivesocket.observable.Disposable;
 

src/main/java/io/reactivesocket/internal/EmptySubscription.java renamed to src/main/java/io/reactivesocket/internal/rx/EmptySubscription.java

@@ -11,7 +11,7 @@
  * the License for the specific language governing permissions and limitations under the License.
  */
 
-package io.reactivesocket.internal;
+package io.reactivesocket.internal.rx;
 
 import org.reactivestreams.*;
 

src/main/java/io/reactivesocket/internal/rx/SubscriptionArbiter.java

@@ -30,9 +30,6 @@
 
 import org.reactivestreams.Subscription;
 
-import io.reactivesocket.internal.BackpressureHelper;
-import io.reactivesocket.internal.SubscriptionHelper;
-
 /**
  * Arbitrates requests and cancellation between Subscriptions.
  */

src/main/java/io/reactivesocket/internal/rx/SubscriptionHelper.java

@@ -0,0 +1,79 @@
+/**
+ * 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.reactivesocket.internal.rx;
+
+import org.reactivestreams.*;
+
+public enum SubscriptionHelper {
+    ;
+    
+    public static boolean validateSubscription(Subscription current, Subscription next) {
+        if (next == null) {
+            return true;
+        }
+        if (current != null) {
+            next.cancel();
+            return true;
+        }
+        return false;
+    }
+    
+    /**
+     * <p>
+     * Make sure error reporting via s.onError is serialized.
+     * 
+     * @param current
+     * @param next
+     * @param s
+     * @return
+     */
+    public static boolean validateSubscription(Subscription current, Subscription next, Subscriber<?> s) {
+        if (next == null) {
+            s.onError(new NullPointerException("next is null"));
+            return true;
+        }
+        if (current != null) {
+            next.cancel();
+            return true;
+        }
+        return false;
+    }
+
+    public static boolean validateRequest(long n) {
+        if (n <= 0) {
+            return true;
+        }
+        return false;
+    }
+    
+    /**
+     * <p>
+     * Make sure error reporting via s.onError is serialized.
+     * 
+     * @param n
+     * @param current
+     * @param s
+     * @return
+     */
+    public static boolean validateRequest(long n, Subscription current, Subscriber<?> s) {
+        if (n <= 0) {
+            if (current != null) {
+                current.cancel();
+            }
+            s.onError(new IllegalArgumentException("n > 0 required but it was " + n));
+            return true;
+        }
+        return false;
+    }
+}