Merge pull request #18 from lwahlmeier/wsServer

Ws server

Author: lwahlmeier

build-all.sh

@@ -3,7 +3,7 @@ cd protocol
 ./gradlew clean build install
 cd ..
 cd server
-./gradlew clean build install --offline
+./gradlew clean build install
 cd ..
 cd client
 ./gradlew clean build install

client/build.shared

@@ -19,10 +19,7 @@ dependencies {
   compile (
     "org.threadly:threadly:$threadlyVersion",
     "org.threadly:litesockets:$litesocketsVersion",
-    "org.threadly:litesockets-http-protocol:$version",
-  )
-  testCompile (
-    "org.threadly:litesockets-http-server:$version",
+    "org.threadly:litesockets-http-protocol:$version"
   )
 }
 

client/gradle/wrapper/gradle-wrapper.properties

@@ -3,4 +3,4 @@ distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-3.3-bin.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-4.4-bin.zip

client/src/main/java/org/threadly/litesockets/client/http/HTTPClient.java

@@ -37,6 +37,7 @@
 import org.threadly.litesockets.protocols.http.shared.HTTPParsingException;
 import org.threadly.litesockets.protocols.http.shared.HTTPRequestType;
 import org.threadly.litesockets.protocols.http.shared.HTTPResponseCode;
+import org.threadly.litesockets.protocols.ws.WebSocketFrameParser.WebSocketFrame;
 import org.threadly.litesockets.utils.IOUtils;
 import org.threadly.litesockets.utils.SSLUtils;
 import org.threadly.util.AbstractService;
@@ -514,7 +515,7 @@ private class HTTPRequestWrapper implements HTTPResponseCallback {
     private long lastRead = Clock.lastKnownForwardProgressingMillis();
 
     public HTTPRequestWrapper(HTTPRequest hr, HTTPAddress ha, ByteBuffer body, long timeout) {
-      hrp.addHTTPRequestCallback(this);
+      hrp.addHTTPResponseCallback(this);
       this.hr = hr;
       this.ha = ha;
       this.body = body;
@@ -545,8 +546,8 @@ public void bodyData(ByteBuffer bb) {
 
     @Override
     public void finished() {
-      slf.setResult(new HTTPResponseData(response, responseMBB.duplicateAndClean()));
-      hrp.removeHTTPRequestCallback(this);
+      slf.setResult(new HTTPResponseData(HTTPClient.this, hr, response, responseMBB.duplicateAndClean()));
+      hrp.removeHTTPResponseCallback(this);
       inProcess.remove(client);
       addBackTCPClient(ha, client);
       processQueue();
@@ -555,6 +556,13 @@ public void finished() {
     @Override
     public void hasError(Throwable t) {
       slf.setFailure(t);
+      client.close();
+    }
+
+    @Override
+    public void websocketData(WebSocketFrame wsf, ByteBuffer bb) {
+      slf.setFailure(new Exception("HTTPClient does not currently support websockets!"));
+      client.close();
     }
   }
 
@@ -563,11 +571,23 @@ public void hasError(Throwable t) {
    */
   public static class HTTPResponseData {
     private final HTTPResponse hr;
+    private final HTTPRequest origRequest;
     private final MergedByteBuffers body;
+    private final HTTPClient client;
 
-    public HTTPResponseData(HTTPResponse hr, MergedByteBuffers bb) {
+    public HTTPResponseData(HTTPClient client, HTTPRequest origRequest, HTTPResponse hr, MergedByteBuffers bb) {
+      this.client = client;
       this.hr = hr;
       this.body = bb;
+      this.origRequest = origRequest;
+    }
+    
+    public HTTPClient getHTTPClient() {
+      return client;
+    }
+    
+    public HTTPRequest getHTTPRequest() {
+      return origRequest;
     }
 
     public HTTPResponse getResponse() {

client/src/main/java/org/threadly/litesockets/client/http/HTTPStreamClient.java

@@ -16,11 +16,11 @@
 import org.threadly.litesockets.SocketExecuter;
 import org.threadly.litesockets.TCPClient;
 import org.threadly.litesockets.protocols.http.request.HTTPRequest;
-import org.threadly.litesockets.protocols.http.request.HTTPRequestBuilder;
 import org.threadly.litesockets.protocols.http.response.HTTPResponse;
 import org.threadly.litesockets.protocols.http.response.HTTPResponseProcessor;
 import org.threadly.litesockets.protocols.http.response.HTTPResponseProcessor.HTTPResponseCallback;
 import org.threadly.litesockets.protocols.http.shared.HTTPUtils;
+import org.threadly.litesockets.protocols.ws.WebSocketFrameParser.WebSocketFrame;
 import org.threadly.litesockets.utils.SSLUtils;
 
 /**
@@ -54,7 +54,6 @@ public class HTTPStreamClient implements StreamingClient {
   private final HTTPResponseProcessor httpProcessor;
 
   private volatile boolean isConnected = false;
-  private volatile boolean headersSent = false;
   private volatile HTTPStreamReader httpReader;
   private volatile SettableListenableFuture<HTTPResponse> slfResponse;
   private volatile HTTPRequest currentHttpRequest;
@@ -65,21 +64,19 @@ public class HTTPStreamClient implements StreamingClient {
    * @param client the {@link TCPClient} to use for this connection.
    * @param headerSent true if the http headers have already been sent, false if they still need to be sent.
    */
-  public HTTPStreamClient(TCPClient client, boolean headerSent) {
-    this(client, client.getRemoteSocketAddress().getHostName(), headerSent);
+  public HTTPStreamClient(TCPClient client) {
+    this(client, client.getRemoteSocketAddress().getHostName());
   }
 
-  public HTTPStreamClient(TCPClient client, String host, boolean headerSent) {
+  public HTTPStreamClient(TCPClient client, String host) {
     this.client = client;
     this.host = host;
-    if(headersSent) {
-      currentHttpRequest = new HTTPRequestBuilder().build();
-    }
     port = client.getRemoteSocketAddress().getPort();
     client.addCloseListener(classCloser);
     httpProcessor = new HTTPResponseProcessor();
-    httpProcessor.addHTTPRequestCallback(requestCB);
-
+    httpProcessor.addHTTPResponseCallback(requestCB);
+    slfResponse = new SettableListenableFuture<HTTPResponse>();
+    isConnected = true;
   }
 
   /**
@@ -99,7 +96,7 @@ public HTTPStreamClient(SocketExecuter se, String host, int port) throws IOExcep
     client.setConnectionTimeout(DEFAULT_TIMEOUT);
     client.addCloseListener(classCloser);
     httpProcessor = new HTTPResponseProcessor();
-    httpProcessor.addHTTPRequestCallback(requestCB);
+    httpProcessor.addHTTPResponseCallback(requestCB);
   }
 
   @Override
@@ -137,6 +134,17 @@ public int getPort() {
     return port;
   }
 
+  @Override
+  public void setRequestResponseHeaders(HTTPRequest httpRequest, HTTPResponse httpResponse, boolean writeResponse) {
+    if(!slfResponse.isDone()) {
+      currentHttpRequest = httpRequest;
+      httpProcessor.processData(httpResponse.getByteBuffer());
+      if(writeResponse) {
+        client.write(httpResponse.getByteBuffer());
+      }
+    }
+  }
+
   /**
    * <p>Tell the client to write an HTTPRequest to the server.  This can technically be done
    * whenever you want to but obviously use only when you know you can sent a request, right after
@@ -169,7 +177,11 @@ public ListenableFuture<?> write(ByteBuffer bb) {
       return client.write(bb);
     }
   }
-
+  
+  public ListenableFuture<?> getLastWriteFuture() {
+    return client.lastWriteFuture();
+  }
+  
   /**
    * Sets the HTTPStreamReader for this client.
    * 
@@ -274,6 +286,13 @@ public void hasError(Throwable t) {
       slfResponse.setFailure(t);
       client.close();
     }
+
+    @Override
+    public void websocketData(WebSocketFrame wsf, ByteBuffer bb) {
+      if(httpReader != null) {
+        httpReader.handle(bb);
+      }
+    }
   }
 
   /**

client/src/main/java/org/threadly/litesockets/client/http/StreamingClient.java

@@ -6,6 +6,8 @@
 import javax.net.ssl.SSLEngine;
 
 import org.threadly.concurrent.future.ListenableFuture;
+import org.threadly.litesockets.protocols.http.request.HTTPRequest;
+import org.threadly.litesockets.protocols.http.response.HTTPResponse;
 
 
 /**
@@ -37,6 +39,16 @@ public interface StreamingClient {
    */
   public void setConnectionTimeout(int timeout);
 
+  /**
+   * Sets the HTTPRequest and HTTPResponse allowing the streaming client to start streaming
+   * w/o sending these in itself.  This is used for the server side of streaming.
+   * 
+   * @param httpRequest
+   * @param httpResponse
+   * @param writeResponse
+   */
+  public void setRequestResponseHeaders(HTTPRequest httpRequest, HTTPResponse httpResponse, boolean writeResponse);
+  
   /**
    * This performs a write to the connection.
    * 
@@ -45,6 +57,8 @@ public interface StreamingClient {
    */
   public ListenableFuture<?> write(ByteBuffer bb);
 
+  public ListenableFuture<?> getLastWriteFuture();
+  
   /**
    * This is called to connect this client to the server.  
    * This will also send in the HTTP upgrade request once the TCP connection is finished.

client/src/main/java/org/threadly/litesockets/client/ws/WebSocketClient.java

@@ -18,9 +18,12 @@
 import org.threadly.litesockets.client.http.HTTPStreamClient;
 import org.threadly.litesockets.client.http.HTTPStreamClient.HTTPStreamReader;
 import org.threadly.litesockets.client.http.StreamingClient;
+import org.threadly.litesockets.protocols.http.request.HTTPRequest;
 import org.threadly.litesockets.protocols.http.request.HTTPRequestBuilder;
 import org.threadly.litesockets.protocols.http.response.HTTPResponse;
+import org.threadly.litesockets.protocols.http.response.HTTPResponseBuilder;
 import org.threadly.litesockets.protocols.http.shared.HTTPConstants;
+import org.threadly.litesockets.protocols.http.shared.HTTPRequestType;
 import org.threadly.litesockets.protocols.http.shared.HTTPResponseCode;
 import org.threadly.litesockets.protocols.ws.WebSocketFrameParser;
 import org.threadly.litesockets.protocols.ws.WebSocketFrameParser.WebSocketFrame;
@@ -34,7 +37,20 @@
  * @author lwahlmeier
  *
  */
-public class WebSocketClient implements StreamingClient{
+public class WebSocketClient implements StreamingClient {
+  public static final HTTPResponse DEFAULT_WS_RESPONSE = new HTTPResponseBuilder()
+      .setResponseCode(HTTPResponseCode.SwitchingProtocols)
+      .setHeader(HTTPConstants.HTTP_KEY_UPGRADE, "websocket")
+      .setHeader(HTTPConstants.HTTP_KEY_CONNECTION, "Upgrade")
+      .setHeader(HTTPConstants.HTTP_KEY_WEBSOCKET_ACCEPT, "123456")
+      .build();
+  public static final HTTPRequest DEFAULT_WS_REQUEST = new HTTPRequestBuilder()
+      .setRequestType(HTTPRequestType.GET)
+      .setHeader(HTTPConstants.HTTP_KEY_UPGRADE, "websocket")
+      .setHeader(HTTPConstants.HTTP_KEY_CONNECTION, "Upgrade")
+      .setHeader(HTTPConstants.HTTP_KEY_WEBSOCKET_VERSION, "13")
+      .setHeader(HTTPConstants.HTTP_KEY_WEBSOCKET_KEY, "")
+      .build(); 
   public static final String WSS_STRING = "wss";
   public static final String WS_STRING = "ws";
   public static final int WSS_PORT = 443;
@@ -57,17 +73,13 @@ public class WebSocketClient implements StreamingClient{
    * @param client the TCPClient to use for this connection.
    * @param alreadyUpgraded true if the connection has already upgraded to do websockets false if the http upgrade is still required.
    */
-  public WebSocketClient(final TCPClient client, final boolean alreadyUpgraded) {
+  public WebSocketClient(final TCPClient client) {
     if(client.isClosed()) {
       throw new IllegalStateException("TCPClient is closed! Can only use an Open TCPClient");
     }
 
-    hsc = new HTTPStreamClient(client, alreadyUpgraded);
-    
-    if(alreadyUpgraded) {
-      sentRequest.set(true);
-      connectFuture.setResult(true);
-    }
+    hsc = new HTTPStreamClient(client);
+    connectFuture.setResult(true);
   }
 
   /**
@@ -266,6 +278,13 @@ public void setWebSocketDataReader(final WebSocketDataReader reader) {
     hsc.setHTTPStreamReader(lsr);
   }
 
+
+  @Override
+  public void setRequestResponseHeaders(HTTPRequest httpRequest, HTTPResponse httpResponse, boolean writeRequest) {
+    hsc.setRequestResponseHeaders(httpRequest, httpResponse, writeRequest);
+    sentRequest.set(true);
+  }
+  
   @Override
   public ListenableFuture<?> write(final ByteBuffer bb) {
     return write(bb, this.wsoc.getValue(), defaultMask);
@@ -302,6 +321,10 @@ public ListenableFuture<?> write(final ByteBuffer bb, final byte opCode, final b
       throw new IllegalStateException("Must be connected first!");
     }
   }
+  
+  public ListenableFuture<?> getLastWriteFuture() {
+    return hsc.getLastWriteFuture();
+  }
 
   @Override
   public ListenableFuture<Boolean> connect() {
@@ -407,4 +430,5 @@ public interface WebSocketDataReader {
     public void onData(WebSocketFrame wsf, ByteBuffer bb);
   }
 
+
 }