Add BigDecimal, LocalDate and LocalDateTime Scala codecs

build.sbt

@@ -1,5 +1,5 @@
 name := "driver-scala-codecs"
-version := "1.0"
+version := "1.1"
 organizationName := "DataStax"
 startYear := Some(2017)
 licenses += ("Apache-2.0", new URL("https://www.apache.org/licenses/LICENSE-2.0.txt"))
@@ -11,7 +11,7 @@ libraryDependencies ++= Seq(
   "com.datastax.cassandra" % "cassandra-driver-core" % "3.3.0",
   "com.datastax.cassandra" % "cassandra-driver-extras" % "3.3.0" % "optional",
   "ch.qos.logback" % "logback-classic" % "1.2.3" % "runtime",
-   "org.scalatest" %% "scalatest" % "3.0.1" % "test",
+  "org.scalatest" %% "scalatest" % "3.0.1" % "test",
   "org.scalacheck" % "scalacheck_2.12" % "1.13.5" % "test"
 )
 

src/main/scala/com/datastax/driver/extras/codecs/scala/BigDecimalCodec2.scala

@@ -0,0 +1,85 @@
+package com.datastax.driver.extras.codecs.scala
+
+import java.nio.ByteBuffer
+
+import com.datastax.driver.core.exceptions.InvalidTypeException
+import com.datastax.driver.core.DataType
+import com.datastax.driver.core.ProtocolVersion
+import com.datastax.driver.core.TypeCodec
+import com.google.common.reflect.TypeToken
+
+import scala.util.Success
+import scala.util.Try
+
+object BigDecimalCodec2
+  extends TypeCodec[BigDecimal](
+    DataType.decimal(),
+    TypeToken.of(classOf[BigDecimal]).wrap()
+  ) with VersionAgnostic[BigDecimal] {
+
+  override def serialize(
+    bigDecimal: BigDecimal,
+    protocolVersion: ProtocolVersion
+  ): ByteBuffer =
+    Option(bigDecimal) match {
+      case Some(value) =>
+        val bigInteger = value.bigDecimal.unscaledValue()
+        val scale = value.scale
+        val bigIntegerBytes = bigInteger.toByteArray
+
+        val bytes = ByteBuffer.allocate(4 + bigIntegerBytes.length)
+        bytes.putInt(scale)
+        bytes.put(bigIntegerBytes)
+        bytes.rewind
+        bytes
+
+      // It is used `null` due to serialize requirement
+      case _ => null // scalastyle:ignore
+    }
+
+  override def deserialize(
+    bytes: ByteBuffer,
+    protocolVersion: ProtocolVersion
+  ): BigDecimal =
+    Option(bytes) match {
+      case Some(value) if value.remaining >= 4 =>
+        val byteBuffer = bytes.duplicate
+        val scale = byteBuffer.getInt
+        val byteArray = new Array[Byte](byteBuffer.remaining)
+        byteBuffer.get(byteArray)
+        BigDecimal(BigInt(byteArray), scale)
+
+      case Some(value) if value.remaining < 4 => throw new InvalidTypeException(
+        s"Invalid decimal value, expecting at least 4 bytes but got ${bytes.remaining}" // scalastyle:ignore
+      )
+
+      // It is used `null` due to deserialize requirement
+      case Some(value) if value.remaining == 0 => null // scalastyle:ignore
+      case _ => null // scalastyle:ignore
+    }
+
+  override def format(bigDecimal: BigDecimal): String =
+    Option(bigDecimal) match {
+      case Some(value) => value.toString()
+      case _ => "NULL"
+    }
+
+  override def parse(bigDecimal: String): BigDecimal = {
+
+    def parseDecimal: BigDecimal =
+      Try(BigDecimal(bigDecimal)) match {
+        case Success(value) => value
+        case _ => throw new IllegalArgumentException(
+          s"Cannot parse decimal value from $bigDecimal"
+        )
+      }
+
+    Option(bigDecimal) match {
+      // It is used `null` due to parse requirement
+      case Some(value)
+        if value.isEmpty || value.equalsIgnoreCase("NULL") || value == null => null // scalastyle:ignore
+      case Some(value) if value.nonEmpty => parseDecimal
+    }
+  }
+
+}

src/main/scala/com/datastax/driver/extras/codecs/scala/CassandraCodecs.scala

@@ -0,0 +1,19 @@
+package com.datastax.driver.extras.codecs.scala
+
+import java.time.LocalDate
+import java.time.LocalDateTime
+
+import com.datastax.driver.core._
+
+trait CassandraCodecs {
+
+  val bigDecimalCodec: TypeCodec[BigDecimal] = BigDecimalCodec2
+  val localDateCodec: TypeCodec[LocalDate] = LocalDateCodec
+  val localDateTimeCodec: TypeCodec[LocalDateTime] = LocalDateTimeCodec
+
+  def registerCodecs(cassandraSession: Session): CodecRegistry = {
+    cassandraSession.getCluster.getConfiguration.getCodecRegistry
+      .register(bigDecimalCodec, localDateCodec, localDateTimeCodec)
+  }
+
+}

src/main/scala/com/datastax/driver/extras/codecs/scala/LocalDateCodec.scala

@@ -0,0 +1,84 @@
+package com.datastax.driver.extras.codecs.scala
+
+import java.nio.ByteBuffer
+import java.time.LocalDate
+import java.time.format.DateTimeFormatter
+import java.time.temporal.ChronoUnit
+
+import com.datastax.driver.core.exceptions.InvalidTypeException
+import com.datastax.driver.core.CodecUtils
+import com.datastax.driver.core.DataType
+import com.datastax.driver.core.ProtocolVersion
+import com.datastax.driver.core.TypeCodec
+import com.google.common.reflect.TypeToken
+
+import scala.util.Success
+import scala.util.Try
+
+object LocalDateCodec
+  extends TypeCodec[LocalDate](
+    DataType.date(),
+    TypeToken.of(classOf[LocalDate]).wrap()
+  ) with VersionAgnostic[LocalDate] {
+
+  override def serialize(
+    localDate: LocalDate,
+    protocolVersion: ProtocolVersion
+  ): ByteBuffer =
+    Option(localDate) match {
+      case Some(value) =>
+        val epoch = LocalDate.ofEpochDay(0)
+        val daysSinceEpoch = ChronoUnit.DAYS.between(epoch, value)
+        val unsigned = CodecUtils.fromSignedToUnsignedInt(daysSinceEpoch.toInt)
+        ByteBuffer.allocate(4).putInt(0, unsigned)
+
+      // It is used `null` due to serialize requirement
+      case _ => null // scalastyle:ignore
+    }
+
+  override def deserialize(
+    bytes: ByteBuffer,
+    protocolVersion: ProtocolVersion
+  ): LocalDate =
+    Option(bytes) match {
+      case Some(value) if value.remaining == 4 =>
+        val unsigned = value.getInt(value.position)
+        val daysSinceEpoch = CodecUtils.fromUnsignedToSignedInt(unsigned)
+        LocalDate.ofEpochDay(daysSinceEpoch)
+
+      case Some(value) if value.remaining == 0 || value.remaining != 4 =>
+        throw new InvalidTypeException(
+          s"Invalid 32-bits integer value, expecting 4 bytes but got ${bytes.remaining}" // scalastyle:ignore
+        )
+
+      // It is used `null` due to deserialize requirement
+      case _ => null // scalastyle:ignore
+    }
+
+  override def format(localDate: LocalDate): String =
+    Option(localDate) match {
+      case Some(value) => value.format(DateTimeFormatter.ISO_LOCAL_DATE)
+      case _ => "NULL"
+    }
+
+  override def parse(localDate: String): LocalDate = {
+
+    def parseDate: LocalDate =
+      Try(LocalDate.parse(localDate, DateTimeFormatter.ISO_LOCAL_DATE)) match {
+        case Success(value) => value
+        case _ => throw new IllegalArgumentException(
+          s"Illegal date format $localDate"
+        )
+      }
+
+    Option(localDate) match {
+      case Some(value) if value.nonEmpty => parseDate
+
+      // It is used `null` due to parse requirement
+      case Some(value) if value.isEmpty || value.equalsIgnoreCase("NULL") => null // scalastyle:ignore
+      case _ => null // scalastyle:ignore
+    }
+
+  }
+
+}

src/main/scala/com/datastax/driver/extras/codecs/scala/LocalDateTimeCodec.scala

@@ -0,0 +1,83 @@
+package com.datastax.driver.extras.codecs.scala
+
+import java.nio.ByteBuffer
+import java.time.format.DateTimeFormatter
+import java.time.Instant
+import java.time.LocalDateTime
+import java.time.ZoneOffset
+
+import com.datastax.driver.core.exceptions.InvalidTypeException
+import com.datastax.driver.core.DataType
+import com.datastax.driver.core.ProtocolVersion
+import com.datastax.driver.core.TypeCodec
+import com.google.common.reflect.TypeToken
+
+import scala.util.Success
+import scala.util.Try
+
+object LocalDateTimeCodec
+  extends TypeCodec[LocalDateTime](
+    DataType.timestamp(),
+    TypeToken.of(classOf[LocalDateTime]).wrap()
+  ) with VersionAgnostic[LocalDateTime] {
+
+  override def serialize(
+    localDateTime: LocalDateTime,
+    protocolVersion: ProtocolVersion
+  ): ByteBuffer =
+    Option(localDateTime) match {
+      case Some(value) =>
+        val milliseconds = value.atZone(ZoneOffset.UTC).toInstant.toEpochMilli
+        ByteBuffer.allocate(8).putLong(0, milliseconds)
+
+      // It is used `null` due to serialize requirement
+      case _ => null // scalastyle:ignore
+    }
+
+  override def deserialize(
+    bytes: ByteBuffer,
+    protocolVersion: ProtocolVersion
+  ): LocalDateTime =
+    Option(bytes) match {
+      case Some(value) if value.remaining == 8 =>
+        val milliseconds = value.getLong(value.position())
+        LocalDateTime.ofInstant(
+          Instant.ofEpochMilli(milliseconds),
+          ZoneOffset.UTC
+        )
+
+      case _ => throw new InvalidTypeException(
+        s"Invalid 64-bits long value, expecting 8 bytes but got ${bytes.remaining}" // scalastyle:ignore
+      )
+    }
+
+  override def format(localDateTime: LocalDateTime): String =
+    Option(localDateTime) match {
+      case Some(value) => value.format(DateTimeFormatter.ISO_LOCAL_DATE_TIME)
+      case _ => "NULL"
+    }
+
+  override def parse(localDateTime: String): LocalDateTime = {
+
+    def parseDateTime: LocalDateTime =
+      Try(LocalDateTime.parse(
+        localDateTime,
+        DateTimeFormatter.ISO_LOCAL_DATE_TIME
+      )) match {
+        case Success(value) => value
+        case _ => throw new IllegalArgumentException(
+          s"Illegal datetime format $localDateTime"
+        )
+      }
+
+    Option(localDateTime) match {
+      case Some(value) if value.nonEmpty => parseDateTime
+
+      // It is used `null` due to parse requirement
+      case Some(value) if value.isEmpty || value.equalsIgnoreCase("NULL") => null // scalastyle:ignore
+      case _ => null // scalastyle:ignore
+    }
+
+  }
+
+}

src/main/scala/com/datastax/driver/extras/codecs/scala/TypeConversions.scala

@@ -23,7 +23,7 @@ import java.util.{Date, UUID}
 
 import com.datastax.driver.core.exceptions.CodecNotFoundException
 import com.datastax.driver.core.{Duration, TypeCodec}
-import com.datastax.driver.extras.codecs.jdk8.{InstantCodec, LocalDateCodec, LocalTimeCodec}
+import com.datastax.driver.extras.codecs.jdk8.{InstantCodec, LocalTimeCodec}
 import com.google.common.reflect.TypeToken
 
 import scala.reflect.runtime.universe._
@@ -45,7 +45,7 @@ object TypeConversions {
       case t if t =:= typeOf[Double] => DoubleCodec
 
       case t if t =:= typeOf[BigInt] => BigIntCodec
-      case t if t =:= typeOf[BigDecimal] => BigDecimalCodec
+      case t if t =:= typeOf[BigDecimal] => BigDecimalCodec2
 
       case t if t =:= typeOf[String] => TypeCodec.varchar()
       case t if t =:= typeOf[ByteBuffer] => TypeCodec.blob()
@@ -56,8 +56,9 @@ object TypeConversions {
       case t if t =:= typeOf[UUID] => TypeCodec.uuid()
 
       case t if t =:= typeOf[java.time.Instant] => InstantCodec.instance
-      case t if t =:= typeOf[java.time.LocalDate] => LocalDateCodec.instance
+      case t if t =:= typeOf[java.time.LocalDate] => LocalDateCodec
       case t if t =:= typeOf[java.time.LocalTime] => LocalTimeCodec.instance
+      case t if t =:= typeOf[java.time.LocalDateTime] => LocalDateTimeCodec
 
       case t if t <:< typeOf[Option[_]] => OptionCodec(toCodec[Any](tpe.typeArgs.head))