Issue 45: Split serializers into format specific libraries

build.gradle

@@ -215,17 +215,117 @@ def getProjectVersion() {
     return ver
 }
 
-project('serializers') {
+project('serializers:shared') {
     dependencies {
         compile project(':common')
         compile project(':client')
+        compile group: 'io.pravega', name: 'pravega-client', version: pravegaVersion
+        compile group: 'org.xerial.snappy', name: 'snappy-java', version: snappyVersion
+        testCompile group: 'org.slf4j', name: 'log4j-over-slf4j', version: slf4jApiVersion
+        testCompile group: 'ch.qos.logback', name: 'logback-classic', version: qosLogbackVersion
+        testCompile group: 'io.pravega', name: 'pravega-test-testcommon', version: pravegaVersion
+    }
+
+    javadoc {
+        title = "Serializers shared"
+        dependsOn delombok
+        source = delombok.outputDir
+        failOnError = true
+        exclude "**/impl/**";
+        options.addBooleanOption("Xdoclint:all,-reference", true)
+    }
+}
+
+project('serializers:avro') {
+    dependencies {
+        compile project(':serializers:shared')
         compile group: 'org.apache.avro', name: 'avro', version: avroVersion
+        testCompile project(path:':serializers:shared', configuration:'testRuntime')
+        testCompile group: 'org.slf4j', name: 'log4j-over-slf4j', version: slf4jApiVersion
+        testCompile group: 'ch.qos.logback', name: 'logback-classic', version: qosLogbackVersion
+        testCompile group: 'io.pravega', name: 'pravega-test-testcommon', version: pravegaVersion
+    }
+
+    javadoc {
+        title = "Serializers avro"
+        dependsOn delombok
+        source = delombok.outputDir
+        failOnError = true
+        exclude "**/impl/**";
+        options.addBooleanOption("Xdoclint:all,-reference", true)
+    }
+
+    jar {
+        from {
+            configurations.compile.collect { it.isDirectory() ? it : zipTree(it) }
+        }
+    }
+}
+
+project('serializers:protobuf') {
+    dependencies {
+        compile project(':serializers:shared')
         compile group: 'com.google.protobuf', name: 'protobuf-java', version: protobufProtocVersion
         compile group: 'com.google.protobuf', name: 'protobuf-java-util', version: protobufUtilVersion
-        compile group: 'io.pravega', name: 'pravega-client', version: pravegaVersion
-        compile group: 'org.xerial.snappy', name: 'snappy-java', version: snappyVersion
-        compile group: 'com.fasterxml.jackson.module', name: 'jackson-module-jsonSchema', version: jacksonVersion
+        testCompile project(path:':serializers:shared', configuration:'testRuntime')
+        testCompile group: 'org.slf4j', name: 'log4j-over-slf4j', version: slf4jApiVersion
+        testCompile group: 'ch.qos.logback', name: 'logback-classic', version: qosLogbackVersion
+        testCompile group: 'io.pravega', name: 'pravega-test-testcommon', version: pravegaVersion
+    }
+
+    javadoc {
+        title = "Serializers protobuf"
+        dependsOn delombok
+        source = delombok.outputDir
+        failOnError = true
+        exclude "**/impl/**";
+        options.addBooleanOption("Xdoclint:all,-reference", true)
+    }
+
+    jar {
+        from {
+            configurations.compile.collect { it.isDirectory() ? it : zipTree(it) }
+        }
+    }
+}
+
+project('serializers:json') {
+    dependencies {
+        compile project(':serializers:shared')
         compile group: 'com.github.everit-org.json-schema', name: 'org.everit.json.schema', version: everitVersion
+        compile group: 'com.fasterxml.jackson.module', name: 'jackson-module-jsonSchema', version: jacksonVersion
+        testCompile project(path:':serializers:shared', configuration:'testRuntime')
+        testCompile group: 'org.slf4j', name: 'log4j-over-slf4j', version: slf4jApiVersion
+        testCompile group: 'ch.qos.logback', name: 'logback-classic', version: qosLogbackVersion
+        testCompile group: 'io.pravega', name: 'pravega-test-testcommon', version: pravegaVersion
+    }
+
+    javadoc {
+        title = "Serializers json"
+        dependsOn delombok
+        source = delombok.outputDir
+        failOnError = true
+        exclude "**/impl/**";
+        options.addBooleanOption("Xdoclint:all,-reference", true)
+    }
+
+    jar {
+        from {
+            configurations.compile.collect { it.isDirectory() ? it : zipTree(it) }
+        }
+    }
+}
+
+project('serializers') {
+    dependencies {
+        compile project(':serializers:avro')
+        compile project(':serializers:protobuf')
+        compile project(':serializers:json')
+        compile group: 'org.xerial.snappy', name: 'snappy-java', version: snappyVersion
+        testCompile project(path:':serializers:shared', configuration:'testRuntime')
+        testCompile files(project(':serializers:avro').sourceSets.test.output)
+        testCompile files(project(':serializers:protobuf').sourceSets.test.output)
+        testCompile files(project(':serializers:json').sourceSets.test.output)
         testCompile group: 'org.slf4j', name: 'log4j-over-slf4j', version: slf4jApiVersion
         testCompile group: 'ch.qos.logback', name: 'logback-classic', version: qosLogbackVersion
         testCompile group: 'io.pravega', name: 'pravega-test-testcommon', version: pravegaVersion
@@ -242,6 +342,8 @@ project('serializers') {
 
     jar {
         manifest {}
+        // add zip64=true so that we are able to pack more than 65k files in the jar. 
+        zip64=true
 
         from {
             configurations.compile.collect { it.isDirectory() ? it : zipTree(it) }
@@ -393,6 +495,10 @@ task publishAllJars() {
     dependsOn ':common:publish'
     dependsOn ':contract:publish'
     dependsOn ':server:publish'
+    dependsOn ':serializers:shared:publish'
+    dependsOn ':serializers:avro:publish'
+    dependsOn ':serializers:json:publish'
+    dependsOn ':serializers:protobuf:publish'
     dependsOn ':serializers:publish'
 }
 

checkstyle/spotbugs-exclude.xml

@@ -6,7 +6,10 @@
         <Package name="io.pravega.schemaregistry.test.integrationtest.generated" />
     </Match>
     <Match> <!-- generated code -->
-        <Package name="io.pravega.schemaregistry.testobjs.generated" />
+        <Package name="io.pravega.schemaregistry.serializer.avro.testobjs.generated" />
+    </Match>
+    <Match> <!-- generated code -->
+        <Package name="io.pravega.schemaregistry.serializer.protobuf.generated" />
     </Match>
     <Match> <!-- does not work well with futures -->
         <Bug pattern="NP_NONNULL_PARAM_VIOLATION" />

gradle/maven.gradle

@@ -39,7 +39,7 @@ plugins.withId('maven') {
                     }
                 }
 
-                pom.artifactId = project.path.replace(':', '')
+                pom.artifactId = project.path.replaceFirst(':', '').replace(':', '-')
                 pom.project {
                     name "Pravega"
                     url "http://pravega.io"
@@ -80,7 +80,7 @@ plugins.withId('maven') {
     install {
         repositories {
             mavenInstaller {
-                pom.artifactId = project.path.replace(':', '')
+                pom.artifactId = project.path.replaceFirst(':', '').replace(':', '-')
             }
         }
     }

serializers/src/main/java/io/pravega/schemaregistry/serializers/AvroDeserializer.java → serializers/avro/src/main/java/io/pravega/schemaregistry/serializer/avro/impl/AvroDeserializer.java

@@ -7,13 +7,16 @@
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  */
-package io.pravega.schemaregistry.serializers;
+package io.pravega.schemaregistry.serializer.avro.impl;
 
 import com.google.common.base.Charsets;
 import com.google.common.base.Preconditions;
+import io.pravega.schemaregistry.serializer.avro.schemas.AvroSchema;
 import io.pravega.schemaregistry.client.SchemaRegistryClient;
 import io.pravega.schemaregistry.contract.data.SchemaInfo;
-import io.pravega.schemaregistry.schemas.AvroSchema;
+import io.pravega.schemaregistry.serializer.shared.impl.AbstractDeserializer;
+import io.pravega.schemaregistry.serializer.shared.impl.EncodingCache;
+import io.pravega.schemaregistry.serializer.shared.impl.SerializerConfig;
 import org.apache.avro.Schema;
 import org.apache.avro.io.BinaryDecoder;
 import org.apache.avro.io.DecoderFactory;
@@ -39,7 +42,7 @@ class AvroDeserializer<T> extends AbstractDeserializer<T> {
     }
 
     @Override
-    protected T deserialize(InputStream inputStream, SchemaInfo writerSchemaInfo, SchemaInfo readerSchemaInfo) throws IOException {
+    public final T deserialize(InputStream inputStream, SchemaInfo writerSchemaInfo, SchemaInfo readerSchemaInfo) throws IOException {
         Preconditions.checkNotNull(writerSchemaInfo);
         Schema writerSchema;
         if (knownSchemas.containsKey(writerSchemaInfo)) {

serializers/src/main/java/io/pravega/schemaregistry/serializers/AvroGenericDeserializer.java → serializers/avro/src/main/java/io/pravega/schemaregistry/serializer/avro/impl/AvroGenericDeserializer.java

@@ -7,12 +7,15 @@
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  */
-package io.pravega.schemaregistry.serializers;
+package io.pravega.schemaregistry.serializer.avro.impl;
 
 import com.google.common.base.Preconditions;
+import io.pravega.schemaregistry.serializer.avro.schemas.AvroSchema;
 import io.pravega.schemaregistry.client.SchemaRegistryClient;
 import io.pravega.schemaregistry.contract.data.SchemaInfo;
-import io.pravega.schemaregistry.schemas.AvroSchema;
+import io.pravega.schemaregistry.serializer.shared.impl.AbstractDeserializer;
+import io.pravega.schemaregistry.serializer.shared.impl.EncodingCache;
+import io.pravega.schemaregistry.serializer.shared.impl.SerializerConfig;
 import org.apache.avro.Schema;
 import org.apache.avro.generic.GenericDatumReader;
 import org.apache.avro.io.BinaryDecoder;
@@ -23,17 +26,17 @@
 import java.io.InputStream;
 import java.util.concurrent.ConcurrentHashMap;
 
-class AvroGenericDeserializer extends AbstractDeserializer<Object> {
+public class AvroGenericDeserializer extends AbstractDeserializer<Object> {
     private final ConcurrentHashMap<SchemaInfo, Schema> knownSchemas;
 
-    AvroGenericDeserializer(String groupId, SchemaRegistryClient client, @Nullable AvroSchema<Object> schema,
+    public AvroGenericDeserializer(String groupId, SchemaRegistryClient client, @Nullable AvroSchema<Object> schema,
                             SerializerConfig.Decoders decoder, EncodingCache encodingCache) {
         super(groupId, client, schema, false, decoder, encodingCache, true);
         this.knownSchemas = new ConcurrentHashMap<>();
     }
 
     @Override
-    protected Object deserialize(InputStream inputStream, SchemaInfo writerSchemaInfo, SchemaInfo readerSchemaInfo) throws IOException {
+    public final Object deserialize(InputStream inputStream, SchemaInfo writerSchemaInfo, SchemaInfo readerSchemaInfo) throws IOException {
         Preconditions.checkNotNull(writerSchemaInfo);
         Schema writerSchema = knownSchemas.computeIfAbsent(writerSchemaInfo, x -> AvroSchema.from(x).getSchema());
         Schema readerSchema = knownSchemas.computeIfAbsent(readerSchemaInfo, x -> AvroSchema.from(x).getSchema());

serializers/src/main/java/io/pravega/schemaregistry/serializers/AvroSerializer.java → serializers/avro/src/main/java/io/pravega/schemaregistry/serializer/avro/impl/AvroSerializer.java

@@ -7,12 +7,13 @@
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  */
-package io.pravega.schemaregistry.serializers;
+package io.pravega.schemaregistry.serializer.avro.impl;
 
+import io.pravega.schemaregistry.serializer.avro.schemas.AvroSchema;
 import io.pravega.schemaregistry.client.SchemaRegistryClient;
-import io.pravega.schemaregistry.codec.Encoder;
+import io.pravega.schemaregistry.serializer.shared.codec.Encoder;
 import io.pravega.schemaregistry.contract.data.SchemaInfo;
-import io.pravega.schemaregistry.schemas.AvroSchema;
+import io.pravega.schemaregistry.serializer.shared.impl.AbstractSerializer;
 import org.apache.avro.Schema;
 import org.apache.avro.generic.GenericDatumWriter;
 import org.apache.avro.generic.IndexedRecord;
@@ -25,9 +26,9 @@
 import java.io.IOException;
 import java.io.OutputStream;
 
-class AvroSerializer<T> extends AbstractSerializer<T> {
+public class AvroSerializer<T> extends AbstractSerializer<T> {
     private final AvroSchema<T> avroSchema;
-    AvroSerializer(String groupId, SchemaRegistryClient client, AvroSchema<T> schema,
+    public AvroSerializer(String groupId, SchemaRegistryClient client, AvroSchema<T> schema,
                    Encoder encoder, boolean registerSchema) {
         super(groupId, client, schema, encoder, registerSchema, true);
         this.avroSchema = schema;