some new functionality and fixes

.gitignore

@@ -7,3 +7,5 @@ release.properties
 pom.xml.releaseBackup
 .DS_Store
 pom.xml.asc
+.gradle
+/build

build.gradle

@@ -0,0 +1,21 @@
+apply plugin: 'java'
+apply plugin: 'maven'
+
+group = GROUP
+version = VERSION
+
+description = "SSE EventSource Java Client"
+
+sourceCompatibility = 1.7
+targetCompatibility = 1.7
+
+repositories {
+     maven { url "http://repo.maven.apache.org/maven2" }
+}
+dependencies {
+    compile group: 'org.slf4j', name: 'slf4j-api', version:'1.7.6'
+    compile group: 'org.jboss.netty', name: 'netty', version:'3.2.4.Final'
+    testCompile group: 'org.webbitserver', name: 'webbit', version:'0.1.13'
+    testCompile group: 'junit', name: 'junit', version:'4.8.2'
+    testCompile group: 'org.mockito', name: 'mockito-all', version:'1.8.5'
+}

gradle.properties

@@ -0,0 +1,22 @@
+# Project-wide Gradle settings.
+
+# IDE (e.g. Android Studio) users:
+# Settings specified in this file will override any Gradle settings
+# configured through the IDE.
+
+# For more details on how to configure your build environment visit
+# http://www.gradle.org/docs/current/userguide/build_environment.html
+
+# Specifies the JVM arguments used for the daemon process.
+# The setting is particularly useful for tweaking memory settings.
+# Default value: -Xmx10248m -XX:MaxPermSize=256m
+# org.gradle.jvmargs=-Xmx2048m -XX:MaxPermSize=512m -XX:+HeapDumpOnOutOfMemoryError -Dfile.encoding=UTF-8
+
+# When configured, Gradle will run in incubating parallel mode.
+# This option should only be used with decoupled projects. More details, visit
+# http://www.gradle.org/docs/current/userguide/multi_project_builds.html#sec:decoupled_projects
+# org.gradle.parallel=true
+
+GROUP=com.github.flocsy
+VERSION=0.2-SNAPSHOT
+ARTIFACT_ID=eventsource-client

pom.xml

@@ -1,36 +1,26 @@
 <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
     <modelVersion>4.0.0</modelVersion>
-    <groupId>com.github.aslakhellesoy</groupId>
+    <groupId>com.github.flocsy</groupId>
     <artifactId>eventsource-client</artifactId>
     <name>${project.artifactId}</name>
     <description>A Java EventSource Client</description>
-    <url>http://aslakhellesoy.github.com/eventsource-java</url>
-    <version>0.1.2.1</version>
+    <url>http://github.com/flocsy/eventsource-java</url>
+    <version>0.2-SNAPSHOT</version>
     <packaging>jar</packaging>
-    <parent>
-        <groupId>org.sonatype.oss</groupId>
-        <artifactId>oss-parent</artifactId>
-        <version>6</version>
-    </parent>
     <licenses>
         <license>
             <name>BSD License</name>
             <url>http://www.opensource.org/licenses/bsd-license</url>
             <distribution>repo</distribution>
         </license>
     </licenses>
-    <scm>
-        <connection>scm:git:git://github.com/aslakhellesoy/eventsource-java.git</connection>
-        <developerConnection>scm:git:[email protected]:aslakhellesoy/eventsource-java.git</developerConnection>
-        <url>git://github.com/aslakhellesoy/eventsource-java.git</url>
-    </scm>
-    <repositories>
-        <repository>
-            <id>repository.jboss.org</id>
-            <url>http://repository.jboss.org/nexus/content/groups/public/</url>
-        </repository>
-    </repositories>
     <dependencies>
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-api</artifactId>
+            <version>1.7.6</version>
+            <scope>compile</scope>
+        </dependency>
         <dependency>
             <groupId>org.jboss.netty</groupId>
             <artifactId>netty</artifactId>
@@ -67,23 +57,18 @@
                     <target>1.6</target>
                 </configuration>
             </plugin>
-            <plugin>
-                <groupId>org.apache.maven.plugins</groupId>
-                <artifactId>maven-gpg-plugin</artifactId>
-                <version>1.1</version>
-                <configuration>
-                    <useAgent>true</useAgent>
-                </configuration>
-                <executions>
-                    <execution>
-                        <id>sign-artifacts</id>
-                        <phase>verify</phase>
-                        <goals>
-                            <goal>sign</goal>
-                        </goals>
-                    </execution>
-                </executions>
-            </plugin>
         </plugins>
     </build>
+    <distributionManagement>
+        <repository>
+            <id>nx-releases</id>
+            <name>Nexus Releases</name>
+            <url>${nexus.releases.url}</url>
+        </repository>
+        <snapshotRepository>
+            <id>nx-snapshots</id>
+            <name>Nexus Snapshots</name>
+            <url>${nexus.snapshots.url}</url>
+        </snapshotRepository>
+    </distributionManagement>
 </project>

src/main/java/com/github/eventsource/client/EventSource.java

@@ -12,13 +12,15 @@
 import org.jboss.netty.handler.codec.frame.Delimiters;
 import org.jboss.netty.handler.codec.http.HttpRequestEncoder;
 import org.jboss.netty.handler.codec.string.StringDecoder;
+import org.jboss.netty.handler.ssl.SslHandler;
 
+import javax.net.ssl.SSLEngine;
 import java.net.InetSocketAddress;
 import java.net.URI;
-import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 
-public class EventSource  {
+public class EventSource implements EventSourceHandler {
     public static final long DEFAULT_RECONNECTION_TIME_MILLIS = 2000;
 
     public static final int CONNECTING = 0;
@@ -27,7 +29,9 @@ public class EventSource  {
 
     private final ClientBootstrap bootstrap;
     private final EventSourceChannelHandler clientHandler;
+    private final EventSourceHandler eventSourceHandler;
 
+    private URI uri;
     private int readyState;
 
     /**
@@ -43,18 +47,37 @@ public class EventSource  {
      * @param eventSourceHandler receives events
      * @see #close()
      */
-    public EventSource(Executor executor, long reconnectionTimeMillis, final URI uri, EventSourceHandler eventSourceHandler) {
+    protected EventSource(ExecutorService executor, long reconnectionTimeMillis, final URI uri, EventSourceHandler eventSourceHandler) {
+        this(executor, reconnectionTimeMillis, uri, null, eventSourceHandler);
+    }
+
+    protected EventSource(ExecutorService executor, long reconnectionTimeMillis, final URI uri, final SSLEngineProvider sslEngineProvider, EventSourceHandler eventSourceHandler) {
+        this.eventSourceHandler = eventSourceHandler;
+
         bootstrap = new ClientBootstrap(
                 new NioClientSocketChannelFactory(
                         Executors.newSingleThreadExecutor(),
                         Executors.newSingleThreadExecutor()));
-        bootstrap.setOption("remoteAddress", new InetSocketAddress(uri.getHost(), uri.getPort()));
+        this.uri = uri;
+
+        bootstrap.setOption("remoteAddress", new InetSocketAddress(uri.getHost(), getPort(uri)));
+
+//        clientHandler = new EventSourceChannelHandler(new AsyncEventSourceHandler(executor, eventSourceHandler), reconnectionTimeMillis, bootstrap, uri);
+        // add this class as the event source handler so the connect() call can be intercepted
+        AsyncEventSourceHandler asyncHandler = new AsyncEventSourceHandler(executor, this);
 
-        clientHandler = new EventSourceChannelHandler(new AsyncEventSourceHandler(executor, eventSourceHandler), reconnectionTimeMillis, bootstrap, uri);
+        clientHandler = new EventSourceChannelHandler(asyncHandler, reconnectionTimeMillis, bootstrap, uri);
 
         bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
             public ChannelPipeline getPipeline() throws Exception {
                 ChannelPipeline pipeline = Channels.pipeline();
+
+                if (uri.getScheme().equals("https") && sslEngineProvider != null) {
+                    SSLEngine engine = sslEngineProvider.createSSLEngine();
+                    engine.setUseClientMode(true);
+                    pipeline.addLast("ssl", new SslHandler(engine));
+                }
+
                 pipeline.addLast("line", new DelimiterBasedFrameDecoder(Integer.MAX_VALUE, Delimiters.lineDelimiter()));
                 pipeline.addLast("string", new StringDecoder());
 
@@ -65,25 +88,56 @@ public ChannelPipeline getPipeline() throws Exception {
         });
     }
 
-    public EventSource(String uri, EventSourceHandler eventSourceHandler) {
-        this(URI.create(uri), eventSourceHandler);
+    protected EventSource(String uri, EventSourceHandler eventSourceHandler) {
+        this(uri, null, eventSourceHandler);
+    }
+
+    protected EventSource(String uri, SSLEngineProvider sslEngineProvider, EventSourceHandler eventSourceHandler) {
+        this(URI.create(uri), sslEngineProvider, eventSourceHandler);
     }
 
-    public EventSource(URI uri, EventSourceHandler eventSourceHandler) {
-        this(Executors.newSingleThreadExecutor(), DEFAULT_RECONNECTION_TIME_MILLIS, uri, eventSourceHandler);
+    protected EventSource(URI uri, EventSourceHandler eventSourceHandler) {
+        this(uri, null, eventSourceHandler);
+    }
+
+    protected EventSource(URI uri, SSLEngineProvider sslEngineProvider, EventSourceHandler eventSourceHandler) {
+        this(Executors.newSingleThreadExecutor(), DEFAULT_RECONNECTION_TIME_MILLIS, uri, sslEngineProvider, eventSourceHandler);
+    }
+
+    /**
+     * Sets a custom HTTP header that will be used when the request is made to establish the SSE channel.
+     *
+     * @param name the HTTP header name
+     * @param value the header value
+     */
+    public void setCustomRequestHeader(String name, String value) {
+        clientHandler.setCustomRequestHeader(name, value);
     }
 
     public ChannelFuture connect() {
         readyState = CONNECTING;
+
+        //To avoid perpetual "SocketUnresolvedException"
+        bootstrap.setOption("remoteAddress", new InetSocketAddress(uri.getHost(), getPort(uri)));
+
         return bootstrap.connect();
     }
 
+    public boolean isConnected() {
+        return (readyState == OPEN);
+    }
+
+    public int getReadyState() {
+        return readyState;
+    }
+
     /**
      * Close the connection
      *
      * @return self
      */
     public EventSource close() {
+        readyState = CLOSED;
         clientHandler.close();
         return this;
     }
@@ -98,4 +152,38 @@ public EventSource join() throws InterruptedException {
         clientHandler.join();
         return this;
     }
+
+    @Override
+    public void onConnect() throws Exception {
+        // flag the connection as open
+        readyState = OPEN;
+
+        // pass event to the proper handler
+        eventSourceHandler.onConnect();
+    }
+
+    @Override
+    public void onMessage(String event, MessageEvent message) throws Exception {
+        // pass event to the proper handler
+        eventSourceHandler.onMessage(event, message);
+    }
+
+    @Override
+    public void onError(Throwable t) {
+        // pass event to the proper handler
+        eventSourceHandler.onError(t);
+    }
+
+    @Override
+    public void onClosed(boolean willReconnect) {
+        eventSourceHandler.onClosed(willReconnect);
+    }
+
+    static public int getPort(URI uri) {
+        int port = uri.getPort();
+        if (port == -1) {
+            port = (uri.getScheme().equals("https")) ? 443 : 80;
+        }
+        return port;
+    }
 }

src/main/java/com/github/eventsource/client/EventSourceHandler.java

@@ -4,4 +4,5 @@ public interface EventSourceHandler {
     void onConnect() throws Exception;
     void onMessage(String event, MessageEvent message) throws Exception;
     void onError(Throwable t);
+    void onClosed(boolean willReconnect);
 }

src/main/java/com/github/eventsource/client/SSLEngineProvider.java

@@ -0,0 +1,7 @@
+package com.github.eventsource.client;
+
+import javax.net.ssl.SSLEngine;
+
+public interface SSLEngineProvider {
+    SSLEngine createSSLEngine();
+}