Add akka-http backend implementation

README.md

@@ -1,7 +1,8 @@
-Java backend for `Socket.IO` library (http://socket.io/)
+Java/Scala backend for `Socket.IO` library (http://socket.io/)
 
 Supports `Socket.IO` clients version 1.0+
-Requires JSR 356-compatible server (tested with Jetty 9 and Tomcat 8)
+
+There is currently support for JSR 356-compatible servlet containers (tested with Jetty 9 and Tomcat 8) and Akka-HTTP.
 
 Right now only websocket and XHR polling transports are implemented.
 
@@ -41,3 +42,21 @@ When Jetty server is embedded into your application, but websocket endpoint is e
                 });
 ```
 See example in [com.codeminders.socketio.sample.jetty.ChatServer](https://github.com/codeminders/socket.io-server-java/blob/master/samples/jetty/src/main/java/com/codeminders/socketio/sample/jetty/ChatServer.java)
+
+## Akka-HTTP usage
+
+The akka-http support revolves around a single exposed class, `SocketIOAkkaHttp`.
+
+```scala
+        val socketIO = SocketIOAkkaHttp().fold(sys.error, identity)
+
+        val routes: Route = {
+            pathPrefix("socket.io") {
+                socketIO.route
+            }
+        }
+
+        Http().bindAndHandle(routes, "localhost", 8080)
+```
+
+The socket URL can be changed by moving the `.route` call to another area in your routes hierarchy.  To configure limits and timeouts, an optional `SocketIOAkkaHttpSettings` instance can be passed to the `SocketIOAkkaHttp()` constructor.

pom.xml

@@ -69,6 +69,14 @@
                 <role>contributor</role>
             </roles>
         </contributor>
+        <contributor>
+            <name>Brian Tarricone</name>
+            <email>[email protected]</email>
+            <timezone>-8</timezone>
+            <roles>
+                <role>contributor</role>
+            </roles>
+        </contributor>
     </contributors>
 
     <licenses>
@@ -94,6 +102,7 @@
         <module>socket-io</module>
         <module>socket-io-servlet</module>
         <module>samples</module>
+        <module>socket-io-akka-http-scala</module>
     </modules>
 
     <distributionManagement>

socket-io-akka-http-scala/pom.xml

@@ -0,0 +1,108 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<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>
+
+    <parent>
+        <artifactId>socketio-parent</artifactId>
+        <groupId>com.codeminders.socketio</groupId>
+        <version>2.0.0-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>socket-io-akka-http-scala_${scala.binary.version}</artifactId>
+    <packaging>jar</packaging>
+
+    <properties>
+        <scala.binary.version>2.12</scala.binary.version>
+        <scala.version>2.12.8</scala.version>
+        <akka.version>2.5.19</akka.version>
+        <akka-http.version>10.1.6</akka-http.version>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>com.codeminders.socketio</groupId>
+            <artifactId>socket-io</artifactId>
+            <version>2.0.0-SNAPSHOT</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.scala-lang</groupId>
+            <artifactId>scala-library</artifactId>
+            <version>${scala.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>com.typesafe.akka</groupId>
+            <artifactId>akka-actor_${scala.binary.version}</artifactId>
+            <version>${akka.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>com.typesafe.akka</groupId>
+            <artifactId>akka-stream_${scala.binary.version}</artifactId>
+            <version>${akka.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>com.typesafe.akka</groupId>
+            <artifactId>akka-http-core_${scala.binary.version}</artifactId>
+            <version>${akka-http.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>com.typesafe.akka</groupId>
+            <artifactId>akka-http_${scala.binary.version}</artifactId>
+            <version>${akka-http.version}</version>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>net.alchim31.maven</groupId>
+                <artifactId>scala-maven-plugin</artifactId>
+                <version>3.4.4</version>
+                <configuration>
+                    <recompileMode>incremental</recompileMode>
+                    <args>
+                        <arg>-deprecation</arg>
+                        <arg>-encoding</arg><arg>utf-8</arg>
+                        <arg>-feature</arg>
+                        <arg>-unchecked</arg>
+                        <arg>-Xlint:infer-any</arg>
+                        <arg>-Xlint:missing-interpolator</arg>
+                        <arg>-Xlint:nullary-unit</arg>
+                        <arg>-Xlint:private-shadow</arg>
+                        <arg>-Xlint:type-parameter-shadow</arg>
+                        <arg>-Xlint:unsound-match</arg>
+                        <arg>-Ypartial-unification</arg>
+                        <arg>-Yno-adapted-args</arg>
+                        <arg>-Ywarn-dead-code</arg>
+                        <arg>-Ywarn-extra-implicit</arg>
+                        <arg>-Ywarn-infer-any</arg>
+                        <arg>-Ywarn-nullary-unit</arg>
+                        <arg>-Ywarn-unused:imports</arg>
+                        <arg>-Ywarn-unused:locals</arg>
+                        <arg>-Ywarn-unused:privates</arg>
+                    </args>
+                    <charset>UTF-8</charset>
+                </configuration>
+                <executions>
+                    <execution>
+                        <id>scala-compile-first</id>
+                        <phase>process-resources</phase>
+                        <goals>
+                            <goal>add-source</goal>
+                            <goal>compile</goal>
+                        </goals>
+                    </execution>
+                    <execution>
+                        <id>scala-test-compile</id>
+                        <phase>process-test-resources</phase>
+                        <goals>
+                            <goal>testCompile</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
+        </plugins>
+    </build>
+</project>

socket-io-akka-http-scala/src/main/resources/reference.conf

@@ -0,0 +1,14 @@
+com.codeminders.socket-io.server.akka-http {
+  # Whether or not to allow all origins via CORS
+  allow-all-origins = true
+  # A list of CORS allowed origins (only used if 'allow-all-origins' is false)
+  allowed-origins = []
+  # Interval at which to expect application-level pings
+  ping-interval = 25 seconds
+  # Time after which a connection will be considered dead if a ping has not been received
+  ping-timeout = 60 seconds
+  # Maximum size of binary WebSocket messages.
+  max-binary-message-size = 8192
+  # Maximum size of text WebSocket messages.
+  max-text-message-size = 32768
+}

socket-io-akka-http-scala/src/main/scala/org/spurint/socketio/server/akkahttp/AkkaHttpRequestWrapper.scala

@@ -0,0 +1,27 @@
+package org.spurint.socketio.server.akkahttp
+
+import akka.http.scaladsl.model.{HttpRequest => AkkaHttpRequest}
+import akka.stream.Materializer
+import akka.stream.scaladsl.{Keep, StreamConverters}
+import com.codeminders.socketio.server.HttpRequest
+import java.io.{BufferedReader, InputStream, InputStreamReader}
+import java.util.Locale
+
+private[akkahttp] class AkkaHttpRequestWrapper(request: AkkaHttpRequest)(implicit materializer: Materializer) extends HttpRequest {
+  private lazy val query = request.uri.query()
+  private lazy val entityInputStream = request.entity.dataBytes.toMat(StreamConverters.asInputStream())(Keep.right).run
+  private lazy val entityReader = new BufferedReader(new InputStreamReader(entityInputStream))
+
+  override def getMethod: String = request.method.value
+
+  override def getHeader(name: String): String =
+    request.headers.find(_.lowercaseName == name.toLowerCase(Locale.US)).map(_.value).orNull
+
+  override def getContentType: String = request.entity.contentType.value
+
+  override def getParameter(name: String): String = query.get(name).orNull
+
+  override def getInputStream: InputStream = entityInputStream
+
+  override def getReader: BufferedReader = entityReader
+}

socket-io-akka-http-scala/src/main/scala/org/spurint/socketio/server/akkahttp/AkkaHttpResponseWrapper.scala

@@ -0,0 +1,60 @@
+package org.spurint.socketio.server.akkahttp
+
+import akka.http.scaladsl.model
+import akka.http.scaladsl.model._
+import akka.http.scaladsl.model.headers.RawHeader
+import akka.stream.scaladsl.StreamConverters
+import com.codeminders.socketio.server.HttpResponse
+import java.io.{OutputStream, PipedInputStream, PipedOutputStream}
+import java.nio.charset.StandardCharsets
+import java.util.concurrent.ConcurrentHashMap
+import scala.collection.JavaConverters._
+
+private[akkahttp] class AkkaHttpResponseWrapper extends HttpResponse with AutoCloseable {
+  @volatile private var statusCode: StatusCode = StatusCodes.OK
+  private val headers = new ConcurrentHashMap[String, String]
+  @volatile private var contentType: Option[String] = None
+
+  private lazy val inputStream = new PipedInputStream()
+  private lazy val outputStream = new PipedOutputStream(inputStream)
+  private lazy val entitySource = StreamConverters.fromInputStream(() => inputStream)
+
+  override def setHeader(name: String, value: String): Unit = headers.put(name, value)
+
+  override def setContentType(contentType: String): Unit = this.contentType = Option(contentType)
+
+  override def getOutputStream: OutputStream = outputStream
+
+  override def sendError(statusCode: Int, message: String): Unit = {
+    this.statusCode = parseStatusCode(statusCode)
+    outputStream.write(message.getBytes(StandardCharsets.UTF_8))
+    close()
+  }
+
+  override def sendError(statusCode: Int): Unit = {
+    this.statusCode = parseStatusCode(statusCode)
+    close()
+  }
+
+  override def flushBuffer(): Unit = outputStream.flush()
+
+  override def close(): Unit = {
+    flushBuffer()
+    outputStream.close()
+  }
+
+  lazy val toAkkaHttpResponse: Either[String, model.HttpResponse] = {
+    close()
+    this.contentType.map(s => ContentType.parse(s)).getOrElse(Right(ContentTypes.NoContentType)).map { contentType =>
+      model.HttpResponse(
+        status = statusCode,
+        headers = headers.asScala.map { case (name, value) => RawHeader(name, value) }.to[scala.collection.immutable.Seq],
+        entity = HttpEntity(contentType, entitySource)
+      )
+    }.swap.map { errors => errors.map(_.summary).mkString("; ") }.swap
+  }
+
+  private def parseStatusCode(statusCode: Int): StatusCode = {
+    StatusCodes.getForKey(statusCode).getOrElse(StatusCodes.custom(statusCode, "", ""))
+  }
+}