[#76] Protocol v3 & v4 support, serialization/deserialization rewrite using scodec

.travis.yml

@@ -8,9 +8,9 @@ sudo: false
 install: ./gradlew assemble -x signArchives
 script:
   - jdk_switcher use openjdk6
-  - ./gradlew check -x signArchives -x java-it-tests:common:check -x java-it-tests:driver20:check -x java-it-tests:driver21:check -x java-it-tests:driver30:check
+  - ./gradlew check -x signArchives -x java-it-tests:common:check -x java-it-tests:common-v3:check -x java-it-tests:driver20:check -x java-it-tests:driver21:check -x java-it-tests:driver30:check
   - jdk_switcher use oraclejdk8
-  - ./gradlew java-it-tests:common:check java-it-tests:driver20:check java-it-tests:driver21:check java-it-tests:driver30:check -x signArchives
+  - ./gradlew java-it-tests:common:check java-it-tests:common-v3:check java-it-tests:driver20:check java-it-tests:driver21:check java-it-tests:driver30:check -x signArchives
 
 cache:
   directories:

README.md

@@ -1,4 +1,4 @@
-# Scassandra - Making testing Cassandra easy
+# Scassandra - Making testing Apache Cassandra easy
 
 
 ## Migrating to Version 1.*
@@ -16,7 +16,9 @@ HTTP JSON API:
 ![TravisCI](https://travis-ci.org/scassandra/scassandra-server.svg?branch=master) 
 [![Gitter](https://badges.gitter.im/Join%20Chat.svg)](https://gitter.im/scassandra/scassandra-server?utm_source=badge&utm_medium=badge&utm_campaign=pr-badge&utm_content=badge)
 
-Datastax Java 2.2, 2.1, 2.0 and 1.0 support. 
+Supports Apache Cassandra Protocol Versions 1 through 4.
+
+Datastax Java 3.x and 2.x support.
 
 Stubbed Cassandra runs as a separate process that your application will believe is a real Cassandra node. 
 It does this by implementing the server side of the binary protocol. 
@@ -38,6 +40,8 @@ To run all tests:
 
 It is made up of:
 
+* cql-antlr - parses cql types using an [antlr4](http://www.antlr.org/) grammar.
+* codec - parses Apache Cassandra's native protocol using [scodec](http://scodec.org/).
 * server - accepts requests over the native protocol and priming over JSON/HTTP. Written in Scala
 * java-client - wrapper for starting the sever in the same JVM as unit/integration tests and a Java API for the priming endpoinds
 * java-it-tests - integration tests against multiple versions of the DataStax Java driver

build.gradle

@@ -28,13 +28,14 @@ subprojects {
         akkaVersion = '2.3.9'
         commonsCodecVersion = '1.10'
         httpClientVersion = '4.3.3'
-        javaDriverVersion = '2.0.10'
+        javaDriverVersion = '3.1.1'
         junitVersion = '4.11'
         guavaVersion = '17.0'
         gsonVersion = '2.5'
         logbackVersion = '1.1.1'
         mockitoVersion = '1.9.5'
-        scalaVersion = '2.11.7'
+        scalaVersion = '2.11.8'
+        scodecVersion = '1.10.2'
         slf4jVersion = '1.7.10'
         sprayVersion = '1.3.3'
         sprayJsonVersion = '1.3.2'

codec/build.gradle

@@ -0,0 +1,137 @@
+buildscript {
+    repositories {
+        mavenCentral()
+        maven {
+            url "https://plugins.gradle.org/m2/"
+        }
+        maven { url "https://oss.sonatype.org/content/groups/public" }
+    }
+    dependencies {
+        classpath "com.github.maiflai:gradle-scalatest:0.7"
+    }
+}
+
+apply plugin: "com.github.maiflai.scalatest"
+apply plugin: 'scala'
+apply plugin: 'signing'
+apply plugin: 'maven'
+
+group 'org.scassandra'
+jar.baseName = 'scassandra-codec_2.11'
+
+if (!project.hasProperty("ossrhUsername")) {
+    ext.ossrhUsername = "dummy"
+}
+if (!project.hasProperty("ossrhPassword")) {
+    ext.ossrhPassword = "dummy"
+}
+
+repositories {
+    mavenCentral()
+    maven {
+        url "https://oss.sonatype.org/content/repositories/snapshots"
+    }
+    maven {
+        url "https://oss.sonatype.org/content/groups/staging"
+    }
+}
+
+compileJava {
+    sourceCompatibility = "1.6"
+    targetCompatibility = "1.6"
+}
+
+
+compileScala {
+    sourceCompatibility = "1.6"
+    targetCompatibility = "1.6"
+}
+
+task sourceJar(type: Jar) {
+    classifier = 'sources'
+    baseName = 'scassandra-codec_2.11'
+    from sourceSets.main.allScala
+}
+
+task javadocJar(type: Jar, dependsOn: javadoc) {
+    classifier = 'javadoc'
+    baseName = 'scassandra-codec'
+    excludes = ['**/**']
+    from 'build/docs/javadoc'
+}
+
+artifacts {
+    archives jar
+    archives javadocJar
+    archives sourceJar
+}
+
+signing {
+    sign configurations.archives
+}
+
+uploadArchives {
+    repositories {
+        mavenDeployer {
+            beforeDeployment { MavenDeployment deployment -> signing.signPom(deployment) }
+
+            repository(url: "https://oss.sonatype.org/service/local/staging/deploy/maven2/") {
+                authentication(userName: ossrhUsername, password: ossrhPassword)
+            }
+
+            snapshotRepository(url: "https://oss.sonatype.org/content/repositories/snapshots") {
+                authentication(userName: ossrhUsername, password: ossrhPassword)
+            }
+            pom.artifactId = 'scassandra-codec_2.11'
+
+            pom.project {
+                name 'Scassandra codec'
+                packaging 'jar'
+                // optionally artifactId can be defined here
+                description 'Scassandra codec'
+                url 'https://github.com/scassandra/'
+
+                scm {
+                    connection 'scm:git:[email protected]:scassandra/scassandra-server.git'
+                    developerConnection '[email protected]:scassandra/scassandra-server.git'
+                    url 'https://github.com/scassandra/scassandra-server'
+                }
+
+                licenses {
+                    license {
+                        name 'The Apache Software License, Version 2.0'
+                        url 'http://www.apache.org/licenses/LICENSE-2.0.txt'
+                    }
+                }
+
+                developers {
+                    developer {
+                        id 'chbatey'
+                        name 'Christopher Batey'
+                        email '[email protected]'
+                    }
+                    developer {
+                        id 'tepafoo'
+                        name 'Dogan Narinc'
+                        email '[email protected]'
+                    }
+                }
+            }
+        }
+    }
+}
+
+dependencies {
+    compile "org.scala-lang:scala-library:$scalaVersion"
+    compile "org.scodec:scodec-core_2.11:$scodecVersion"
+    compile "com.google.guava:guava:$guavaVersion"
+
+    testCompile 'org.scalacheck:scalacheck_2.11:1.12.4'
+    testCompile "org.scalatest:scalatest_2.11:2.2.3"
+
+    testRuntime 'org.pegdown:pegdown:1.1.0'
+}
+
+test {
+    maxParallelForks = 1
+}

codec/src/main/scala/org/scassandra/codec/Frame.scala

@@ -0,0 +1,44 @@
+/*
+ * Copyright (C) 2016 Christopher Batey and Dogan Narinc
+ *
+ * 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 org.scassandra.codec
+
+import scodec.Codec
+import scodec.codecs._
+
+/**
+  * Defines a Frame, the top level payload in the spec.  See section one of the spec.
+  *
+  * A Frame is composed of a header which indicates information about the body, and the message body itself.
+  *
+  * @param header Header of the message
+  * @param message The message itself.
+  */
+case class Frame(
+  header: FrameHeader,
+  message: Message
+)
+
+// TODO: Support warnings, custom payload, and tracing.
+
+object Frame {
+  /**
+    * Codec for [[Frame]], reads header and then uses the [[FrameHeader]] to determine the [[Message]].
+    */
+  implicit val codec: Codec[Frame] = {
+    ("header" | FrameHeader.codec) flatPrepend { header =>
+    ("message" | Message.codec(header.opcode)(header.version.version)).hlist
+  }}.as[Frame]
+}

codec/src/main/scala/org/scassandra/codec/FrameHeader.scala

@@ -0,0 +1,141 @@
+/*
+ * Copyright (C) 2016 Christopher Batey and Dogan Narinc
+ *
+ * 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 org.scassandra.codec
+
+import scodec.Codec
+import scodec.codecs._
+
+/**
+  * Indicates the direction of a frame.
+  */
+sealed trait MessageDirection
+
+/**
+  * Represents the 'request' direction (0).
+  */
+case object Request extends MessageDirection
+
+/**
+  * Represents the 'response' direction (1).
+  */
+case object Response extends MessageDirection
+
+object MessageDirection {
+  /**
+    * 1 bit representation of the direction of a frame.  If 0, it's a request, else response.
+    */
+  implicit val codec: Codec[MessageDirection] = mappedEnum(bool,
+    Request -> false,
+    Response -> true
+  )
+}
+
+/**
+  * Represents the first byte in the frame, which indicates the [[ProtocolVersion]] and [[MessageDirection]]
+  * @param direction direction of the frame
+  * @param version protocol version of the frame
+  */
+sealed case class ProtocolFlags(
+  direction: MessageDirection,
+  version: ProtocolVersion
+) {
+
+  /**
+    * Knowing the direction and version, we can provide a [[Codec]] for parsing the header.  This codec starts by
+    * providing these flags and passes them to [[FrameHeader.withProtocol]] to parse the rest of [[FrameHeader]].
+    */
+  lazy val headerCodec: Codec[FrameHeader] = version.headerCodec(direction)
+}
+
+object ProtocolFlags {
+  /**
+    * Represents the first byte in the frame, with the 1st bit representing the [[MessageDirection]] and the remaining
+    * 7 bits representing the [[ProtocolVersion]].
+    */
+  implicit val codec: Codec[ProtocolFlags] = {
+    ("direction" | MessageDirection.codec) ::
+    ("version"   | ProtocolVersion.codec)
+  }.as[ProtocolFlags]
+}
+
+/**
+  * Indicates what extra content is in the frame beyond the [[Message]].
+  * @param warning Whether or not warnings are present.
+  * @param customPayload Whether or not custom payload is present.
+  * @param tracing Whether or not tracing information is present.
+  * @param compression Whether or not the frame body is compressed.
+  */
+sealed case class HeaderFlags(
+  warning: Boolean = false,
+  customPayload: Boolean = false,
+  tracing: Boolean = false,
+  compression: Boolean = false
+)
+
+/**
+  * A version of [[HeaderFlags]] where all bits are disabled.
+  */
+object EmptyHeaderFlags extends HeaderFlags()
+
+object HeaderFlags {
+  /**
+    * [[Codec]] for parsing [[HeaderFlags]] ignores first 4 bits and parses the remaining into the fields present on
+    * header flags.
+    */
+  implicit val codec: Codec[HeaderFlags] = {
+    ("reserved"      | ignore(4)) ::
+    ("warning"       | bool)      ::
+    ("customPayload" | bool)      ::
+    ("tracing"       | bool)      ::
+    ("compression"   | bool)
+  }.dropUnits.as[HeaderFlags]
+}
+
+/**
+  * Represents the Frame Header of a native protocol [[Frame]].
+  * @param version The version and direction of the frame.
+  * @param flags header flags indicating additional content and whether or not the body is compressed.
+  * @param stream The stream id uniquely identifying a [[Frame]] in flight.
+  * @param opcode distinguishes the [[Message]] contained int he [[Frame]].
+  * @param length The length of the body of the [[Frame]].
+  */
+case class FrameHeader (
+  version: ProtocolFlags,
+  flags: HeaderFlags = EmptyHeaderFlags,
+  stream: Int = 0,
+  opcode: Int = ErrorMessage.opcode,
+  length: Long = 0
+)
+
+object FrameHeader {
+
+  private[codec] def withProtocol(protocol: ProtocolFlags) = {
+    ("flags"   | HeaderFlags.codec)              ::
+    ("stream"  | protocol.version.streamIdCodec) ::
+    ("opcode"  | uint8)                          ::
+    ("length"  | uint32)
+  }
+
+  /**
+    * [[Codec]] that parses [[FrameHeader]] by first inspecting the [[ProtocolFlags]] and then uses the
+    * [[ProtocolVersion]] to determine how to parse the rest of the [[FrameHeader]].
+    */
+  implicit val codec: Codec[FrameHeader] = {
+    ProtocolFlags.codec.consume { protocolFlags =>
+      protocolFlags.headerCodec
+    }(_.version)
+  }
+}