added new optional param classInstance to Serializer and Deserializer helper classes

kafka-0.10.x/src/main/scala/monix/kafka/Deserializer.scala

@@ -26,17 +26,33 @@ import language.existentials
 /** Wraps a Kafka `Deserializer`, provided for
   * convenience, since it can be implicitly fetched
   * from the context.
+  *
+  * @param className is the full package path to the Kafka `Deserializer`
+  *
+  * @param classType is the actual [[Class]] for [[className]]
+  *
+  * @param constructor creates an instance of [[classType]].
+  *        This is defaulted with a `Deserializer.Constructor[A]` function that creates a
+  *        new instance using an assumed empty constructor.
+  *        Supplying this parameter allows for manual provision of the `Deserializer`.
   */
 final case class Deserializer[A](
   className: String,
-  classType: Class[_ <: KafkaDeserializer[A]]) {
+  classType: Class[_ <: KafkaDeserializer[A]],
+  constructor: Deserializer.Constructor[A] = (d: Deserializer[A]) => d.classType.newInstance()) {
 
   /** Creates a new instance. */
   def create(): KafkaDeserializer[A] =
-  classType.newInstance()
+    constructor(this)
 }
 
 object Deserializer {
+
+  /** Alias for the function that provides an instance of
+    * the Kafka `Deserializer`.
+    */
+  type Constructor[A] = (Deserializer[A]) => KafkaDeserializer[A]
+
   implicit val forStrings: Deserializer[String] =
     Deserializer(
       className = "org.apache.kafka.common.serialization.StringDeserializer",

kafka-0.10.x/src/main/scala/monix/kafka/Serializer.scala

@@ -26,17 +26,33 @@ import language.existentials
 /** Wraps a Kafka `Serializer`, provided for
   * convenience, since it can be implicitly fetched
   * from the context.
+  *
+  * @param className is the full package path to the Kafka `Serializer`
+  *
+  * @param classType is the actual [[Class]] for [[className]]
+  *
+  * @param constructor creates an instance of [[classType]].
+  *        This is defaulted with a `Serializer.Constructor[A]` function that creates a
+  *        new instance using an assumed empty constructor.
+  *        Supplying this parameter allows for manual provision of the `Serializer`.
   */
 final case class Serializer[A](
   className: String,
-  classType: Class[_ <: KafkaSerializer[A]]) {
+  classType: Class[_ <: KafkaSerializer[A]],
+  constructor: Serializer.Constructor[A] = (s: Serializer[A]) => s.classType.newInstance()) {
 
   /** Creates a new instance. */
   def create(): KafkaSerializer[A] =
-    classType.newInstance()
+    constructor(this)
 }
 
 object Serializer {
+
+  /** Alias for the function that provides an instance of
+    * the Kafka `Serializer`.
+    */
+  type Constructor[A] = (Serializer[A]) => KafkaSerializer[A]
+
   implicit val forStrings: Serializer[String] =
     Serializer(
       className = "org.apache.kafka.common.serialization.StringSerializer",

kafka-0.8.x/src/main/scala/monix/kafka/Deserializer.scala

@@ -23,22 +23,41 @@ import language.existentials
 /** Wraps a Kafka `Decoder`, provided for
   * convenience, since it can be implicitly fetched
   * from the context.
+  *
+  * @param className is the full package path to the Kafka `Decoder`
+  *
+  * @param classType is the actual [[Class]] for [[className]]
+  *
+  * @param constructor creates an instance of [[classType]].
+  *        This is defaulted with a `Deserializer.Constructor[A]` function that creates a
+  *        new instance using an assumed empty or nullable constructor.
+  *        Supplying this parameter allows for manual provision of the `Decoder`.
   */
 final case class Deserializer[A](
   className: String,
-  classType: Class[_ <: KafkaDecoder[A]]) {
+  classType: Class[_ <: KafkaDecoder[A]],
+  constructor: Deserializer.Constructor[A] = Deserializer.reflectCreate[A] _) {
 
   /** Creates a new instance. */
-  def create(): KafkaDecoder[A] = {
-    val constructor = classType.getDeclaredConstructors()(0)
+  def create(): KafkaDecoder[A] =
+    constructor(this)
+}
+
+object Deserializer {
+
+  /** Alias for the function that provides an instance of
+    * the Kafka `Decoder`.
+    */
+  type Constructor[A] = (Deserializer[A]) => KafkaDecoder[A]
+
+  private def reflectCreate[A](d: Deserializer[A]): KafkaDecoder[A] = {
+    val constructor = d.classType.getDeclaredConstructors()(0)
     constructor.getParameterCount match {
-      case 0 => classType.newInstance()
+      case 0 => d.classType.newInstance()
       case 1 => constructor.newInstance(null).asInstanceOf[KafkaDecoder[A]]
     }
   }
-}
 
-object Deserializer {
   implicit val forStrings: Deserializer[String] =
     Deserializer(
       className = "kafka.serializer.StringDecoder",

kafka-0.8.x/src/main/scala/monix/kafka/Serializer.scala

@@ -24,17 +24,33 @@ import language.existentials
 /** Wraps a Kafka `Serializer`, provided for
   * convenience, since it can be implicitly fetched
   * from the context.
+  *
+  * @param className is the full package path to the Kafka `Serializer`
+  *
+  * @param classType is the actual [[Class]] for [[className]]
+  *
+  * @param constructor creates an instance of [[classType]].
+  *        This is defaulted with a `Serializer.Constructor[A]` function that creates a
+  *        new instance using an assumed empty constructor.
+  *        Supplying this parameter allows for manual provision of the `Serializer`.
   */
 final case class Serializer[A](
   className: String,
-  classType: Class[_ <: KafkaSerializer[A]]) {
+  classType: Class[_ <: KafkaSerializer[A]],
+  constructor: Serializer.Constructor[A] = (s: Serializer[A]) => s.classType.newInstance()) {
 
   /** Creates a new instance. */
   def create(): KafkaSerializer[A] =
-    classType.newInstance()
+    constructor(this)
 }
 
 object Serializer {
+
+  /** Alias for the function that provides an instance of
+    * the Kafka `Serializer`.
+    */
+  type Constructor[A] = (Serializer[A]) => KafkaSerializer[A]
+
   implicit val forStrings: Serializer[String] =
     Serializer(
       className = "org.apache.kafka.common.serialization.StringSerializer",

kafka-0.9.x/src/main/scala/monix/kafka/Deserializer.scala

@@ -24,17 +24,33 @@ import language.existentials
 /** Wraps a Kafka `Deserializer`, provided for
   * convenience, since it can be implicitly fetched
   * from the context.
+  *
+  * @param className is the full package path to the Kafka `Deserializer`
+  *
+  * @param classType is the actual [[Class]] for [[className]]
+  *
+  * @param constructor creates an instance of [[classType]].
+  *        This is defaulted with a `Deserializer.Constructor[A]` function that creates a
+  *        new instance using an assumed empty constructor.
+  *        Supplying this parameter allows for manual provision of the `Deserializer`.
   */
 final case class Deserializer[A](
   className: String,
-  classType: Class[_ <: KafkaDeserializer[A]]) {
+  classType: Class[_ <: KafkaDeserializer[A]],
+  constructor: Deserializer.Constructor[A] = (d: Deserializer[A]) => d.classType.newInstance()) {
 
   /** Creates a new instance. */
   def create(): KafkaDeserializer[A] =
-  classType.newInstance()
+    constructor(this)
 }
 
 object Deserializer {
+
+  /** Alias for the function that provides an instance of
+    * the Kafka `Deserializer`.
+    */
+  type Constructor[A] = (Deserializer[A]) => KafkaDeserializer[A]
+
   implicit val forStrings: Deserializer[String] =
     Deserializer(
       className = "org.apache.kafka.common.serialization.StringDeserializer",

kafka-0.9.x/src/main/scala/monix/kafka/Serializer.scala

@@ -24,17 +24,33 @@ import language.existentials
 /** Wraps a Kafka `Serializer`, provided for
   * convenience, since it can be implicitly fetched
   * from the context.
+  *
+  * @param className is the full package path to the Kafka `Serializer`
+  *
+  * @param classType is the actual [[Class]] for [[className]]
+  *
+  * @param constructor creates an instance of [[classType]].
+  *        This is defaulted with a `Serializer.Constructor[A]` function that creates a
+  *        new instance using an assumed empty constructor.
+  *        Supplying this parameter allows for manual provision of the `Serializer`.
   */
 final case class Serializer[A](
   className: String,
-  classType: Class[_ <: KafkaSerializer[A]]) {
+  classType: Class[_ <: KafkaSerializer[A]],
+  constructor: Serializer.Constructor[A] = (s: Serializer[A]) => s.classType.newInstance()) {
 
   /** Creates a new instance. */
   def create(): KafkaSerializer[A] =
-    classType.newInstance()
+    constructor(this)
 }
 
 object Serializer {
+
+  /** Alias for the function that provides an instance of
+    * the Kafka `Serializer`.
+    */
+  type Constructor[A] = (Serializer[A]) => KafkaSerializer[A]
+
   implicit val forStrings: Serializer[String] =
     Serializer(
       className = "org.apache.kafka.common.serialization.StringSerializer",