Merge pull request #59 from ReactiveSocket/reorg

Rx Concat and Package ReOrg

Author: benjchristensen

src/main/java/io/reactivesocket/ConnectionSetupPayload.java

@@ -15,10 +15,10 @@
  */
 package io.reactivesocket;
 
-import io.reactivesocket.internal.SetupFrameFlyweight;
-
 import java.nio.ByteBuffer;
 
+import io.reactivesocket.internal.frame.SetupFrameFlyweight;
+
 /**
  * Exposed to server for determination of RequestHandler based on mime types and SETUP metadata/data
  */

src/main/java/io/reactivesocket/DuplexConnection.java

@@ -19,7 +19,8 @@
 
 import org.reactivestreams.Publisher;
 
-import io.reactivesocket.observable.Observable;
+import io.reactivesocket.rx.Completable;
+import io.reactivesocket.rx.Observable;
 
 /**
  * Represents a connection with input/output that the protocol uses. 

src/main/java/io/reactivesocket/Frame.java

@@ -16,6 +16,14 @@
 package io.reactivesocket;
 
 import io.reactivesocket.internal.*;
+import io.reactivesocket.internal.frame.ErrorFrameFlyweight;
+import io.reactivesocket.internal.frame.FrameHeaderFlyweight;
+import io.reactivesocket.internal.frame.FramePool;
+import io.reactivesocket.internal.frame.LeaseFrameFlyweight;
+import io.reactivesocket.internal.frame.RequestFrameFlyweight;
+import io.reactivesocket.internal.frame.RequestNFrameFlyweight;
+import io.reactivesocket.internal.frame.SetupFrameFlyweight;
+import io.reactivesocket.internal.frame.UnpooledFrame;
 import uk.co.real_logic.agrona.DirectBuffer;
 import uk.co.real_logic.agrona.MutableDirectBuffer;
 

src/main/java/io/reactivesocket/LeaseGovernor.java

@@ -1,6 +1,8 @@
 package io.reactivesocket;
 
 import io.reactivesocket.internal.Responder;
+import io.reactivesocket.lease.NullLeaseGovernor;
+import io.reactivesocket.lease.UnlimitedLeaseGovernor;
 
 public interface LeaseGovernor {
     public static final LeaseGovernor NULL_LEASE_GOVERNOR = new NullLeaseGovernor();

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

@@ -26,13 +26,14 @@
 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.observable.Disposable;
-import io.reactivesocket.observable.Observable;
-import io.reactivesocket.observable.Observer;
+import io.reactivesocket.internal.rx.CompositeCompletable;
+import io.reactivesocket.internal.rx.CompositeDisposable;
+import io.reactivesocket.rx.Completable;
+import io.reactivesocket.rx.Disposable;
+import io.reactivesocket.rx.Observable;
+import io.reactivesocket.rx.Observer;
 import uk.co.real_logic.agrona.BitUtil;
 
 /**

src/main/java/io/reactivesocket/exceptions/Exceptions.java

@@ -19,7 +19,7 @@
 
 import java.nio.ByteBuffer;
 
-import static io.reactivesocket.internal.ErrorFrameFlyweight.*;
+import static io.reactivesocket.internal.frame.ErrorFrameFlyweight.*;
 import static java.nio.charset.StandardCharsets.UTF_8;
 
 public class Exceptions {

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

@@ -0,0 +1,57 @@
+/**
+ * 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;
+
+import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
+
+import io.reactivesocket.Frame;
+import io.reactivesocket.FrameType;
+import io.reactivesocket.Payload;
+import io.reactivesocket.internal.frame.PayloadFragmenter;
+
+public class FragmentedPublisher implements Publisher<Frame> {
+
+	private final PayloadFragmenter fragmenter = new PayloadFragmenter(Frame.METADATA_MTU, Frame.DATA_MTU);
+	private final Publisher<Payload> responsePublisher;
+	private final int streamId;
+	private final FrameType type;
+
+	public FragmentedPublisher(FrameType type, int streamId, Publisher<Payload> responsePublisher) {
+		this.type = type;
+		this.streamId = streamId;
+		this.responsePublisher = responsePublisher;
+	}
+	
+	@Override
+	public void subscribe(Subscriber<? super Frame> child) {
+		child.onSubscribe(new Subscription() {
+
+			@Override
+			public void request(long n) {
+				// TODO Auto-generated method stub
+				
+			}
+
+			@Override
+			public void cancel() {
+				// TODO Auto-generated method stub
+				
+			}});
+	}
+
+}

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

@@ -29,7 +29,6 @@
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
 
-import io.reactivesocket.Completable;
 import io.reactivesocket.ConnectionSetupPayload;
 import io.reactivesocket.DuplexConnection;
 import io.reactivesocket.Frame;
@@ -38,8 +37,13 @@
 import io.reactivesocket.exceptions.CancelException;
 import io.reactivesocket.exceptions.Exceptions;
 import io.reactivesocket.exceptions.Retryable;
-import io.reactivesocket.observable.Disposable;
-import io.reactivesocket.observable.Observer;
+import io.reactivesocket.internal.frame.RequestFrameFlyweight;
+import io.reactivesocket.internal.rx.BackpressureUtils;
+import io.reactivesocket.internal.rx.EmptyDisposable;
+import io.reactivesocket.internal.rx.EmptySubscription;
+import io.reactivesocket.rx.Completable;
+import io.reactivesocket.rx.Disposable;
+import io.reactivesocket.rx.Observer;
 import uk.co.real_logic.agrona.collections.Int2ObjectHashMap;
 
 /**

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

@@ -28,8 +28,12 @@
 import org.reactivestreams.Subscription;
 
 import io.reactivesocket.exceptions.SetupException;
-import io.reactivesocket.observable.Disposable;
-import io.reactivesocket.observable.Observer;
+import io.reactivesocket.internal.frame.SetupFrameFlyweight;
+import io.reactivesocket.internal.rx.EmptyDisposable;
+import io.reactivesocket.internal.rx.EmptySubscription;
+import io.reactivesocket.rx.Completable;
+import io.reactivesocket.rx.Disposable;
+import io.reactivesocket.rx.Observer;
 import uk.co.real_logic.agrona.collections.Int2ObjectHashMap;
 
 /**
@@ -241,6 +245,7 @@ public void onNext(Frame requestFrame) {
 						final RejectedException exception = new RejectedException("No associated lease");
 						responsePublisher = PublisherUtils.errorFrame(streamId, exception);
 					}
+					
 					connection.addOutput(responsePublisher, new Completable() {
 
 						@Override