forked from azinman/cascal
-
Notifications
You must be signed in to change notification settings - Fork 10
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
4 changed files
with
196 additions
and
1 deletion.
There are no files selected for viewing
103 changes: 103 additions & 0 deletions
103
src/main/scala/com/shorrockin/cascal/serialization/TupleSerializer.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,103 @@ | ||
| package com.shorrockin.cascal.serialization | ||
|
|
||
| import java.nio.ByteBuffer | ||
| import java.util.{Date, UUID} | ||
|
|
||
| object TupleSerializer { | ||
|
|
||
| def extractType[T](bytes: ByteBuffer, mf: Manifest[T]): T = { | ||
| val length = (bytes.get() & 0xFF) << 8 | (bytes.get() & 0xFF) | ||
| val typeBuffer = bytes.duplicate | ||
| typeBuffer.limit(typeBuffer.position + length) | ||
|
|
||
| bytes.position(typeBuffer.position + length + 1) | ||
|
|
||
| val ser = Serializer.Default(mf.erasure) | ||
| ser.fromByteBuffer(typeBuffer).asInstanceOf[T] | ||
| } | ||
|
|
||
| def byteBuffer[T](value: T)(implicit mf: Manifest[T]): ByteBuffer = { | ||
| value match { | ||
| case x: String if mf.erasure == classOf[String] => StringSerializer.toByteBuffer(x) | ||
| case x: UUID if mf.erasure == classOf[UUID] => UUIDSerializer.toByteBuffer(x) | ||
| case x: Int if mf.erasure == classOf[Int] => IntSerializer.toByteBuffer(x) | ||
| case x: Long if mf.erasure == classOf[Long] => LongSerializer.toByteBuffer(x) | ||
| case x: Boolean if mf.erasure == classOf[Boolean] => BooleanSerializer.toByteBuffer(x) | ||
| case x: Float if mf.erasure == classOf[Float] => FloatSerializer.toByteBuffer(x) | ||
| case x: Double if mf.erasure == classOf[Double] => DoubleSerializer.toByteBuffer(x) | ||
| case x: Date if mf.erasure == classOf[Date] => DateSerializer.toByteBuffer(x) | ||
| case None => ByteBuffer.allocate(0) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| class CompositeBuffer(val buffers: ByteBuffer*) { | ||
|
|
||
| val lengthBytesSize = 2 | ||
| val endOfComponentSize = 1 | ||
| val compositeOverheadSize = lengthBytesSize + endOfComponentSize | ||
|
|
||
| def buffer(): ByteBuffer = { | ||
| val buffersSize = buffers.foldLeft(0){(sum, buffer) => sum + buffer.remaining} | ||
| val requiredSize = buffersSize + buffers.size * compositeOverheadSize | ||
| val buffer = ByteBuffer.allocate(requiredSize) | ||
|
|
||
| buffers foreach {buff => | ||
| buffer.putShort(buff.remaining.asInstanceOf[Short]).put(buff).put(0.toByte) | ||
| } | ||
| buffer.rewind | ||
| buffer | ||
| } | ||
| } | ||
|
|
||
| object Tuple2Serializer { | ||
| import TupleSerializer._ | ||
|
|
||
| def toByteBuffer[T1: Manifest, T2: Manifest](tuple: Tuple2[T1, T2]): ByteBuffer = { | ||
| val buffer = new CompositeBuffer(byteBuffer(tuple._1), byteBuffer(tuple._2)) | ||
| buffer.buffer | ||
| } | ||
|
|
||
| def fromByteBuffer[T1, T2](bytes:ByteBuffer, mf1: Manifest[T1], mf2: Manifest[T2]): Tuple2[T1, T2] = { | ||
| (extractType(bytes, mf1), extractType(bytes, mf2)) | ||
| } | ||
| } | ||
|
|
||
| object Tuple3Serializer { | ||
| import TupleSerializer._ | ||
|
|
||
| def toByteBuffer[T1: Manifest, T2: Manifest, T3: Manifest](tuple: Tuple3[T1, T2, T3]): ByteBuffer = { | ||
| val buffer = new CompositeBuffer(byteBuffer(tuple._1), byteBuffer(tuple._2), byteBuffer(tuple._3)) | ||
| buffer.buffer | ||
| } | ||
|
|
||
| def fromByteBuffer[T1, T2, T3](bytes:ByteBuffer, mf1: Manifest[T1], mf2: Manifest[T2], mf3: Manifest[T3]): Tuple3[T1, T2, T3] = { | ||
| (extractType(bytes, mf1), extractType(bytes, mf2), extractType(bytes, mf3)) | ||
| } | ||
| } | ||
|
|
||
| object Tuple4Serializer { | ||
| import TupleSerializer._ | ||
|
|
||
| def toByteBuffer[T1: Manifest, T2: Manifest, T3: Manifest, T4: Manifest](tuple: Tuple4[T1, T2, T3, T4]): ByteBuffer = { | ||
| val buffer = new CompositeBuffer(byteBuffer(tuple._1), byteBuffer(tuple._2), byteBuffer(tuple._3), byteBuffer(tuple._4)) | ||
| buffer.buffer | ||
| } | ||
|
|
||
| def fromByteBuffer[T1, T2, T3, T4](bytes:ByteBuffer, mf1: Manifest[T1], mf2: Manifest[T2], mf3: Manifest[T3], mf4: Manifest[T4]): Tuple4[T1, T2, T3, T4] = { | ||
| (extractType(bytes, mf1), extractType(bytes, mf2), extractType(bytes, mf3), extractType(bytes, mf4)) | ||
| } | ||
| } | ||
|
|
||
| object Tuple5Serializer { | ||
| import TupleSerializer._ | ||
|
|
||
| def toByteBuffer[T1: Manifest, T2: Manifest, T3: Manifest, T4: Manifest, T5: Manifest](tuple: Tuple5[T1, T2, T3, T4, T5]): ByteBuffer = { | ||
| val buffer = new CompositeBuffer(byteBuffer(tuple._1), byteBuffer(tuple._2), byteBuffer(tuple._3), byteBuffer(tuple._4), byteBuffer(tuple._5)) | ||
| buffer.buffer | ||
| } | ||
|
|
||
| def fromByteBuffer[T1, T2, T3, T4, T5](bytes:ByteBuffer, mf1: Manifest[T1], mf2: Manifest[T2], mf3: Manifest[T3], mf4: Manifest[T4], mf5: Manifest[T5]): Tuple5[T1, T2, T3, T4, T5] = { | ||
| (extractType(bytes, mf1), extractType(bytes, mf2), extractType(bytes, mf3), extractType(bytes, mf4), extractType(bytes, mf5)) | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,78 @@ | ||
| package com.shorrockin.cascal | ||
|
|
||
| import org.junit.{Assert, Test} | ||
| import com.shorrockin.cascal.utils.Conversions._ | ||
| import com.shorrockin.cascal.serialization.TupleSerializer | ||
| import Assert._ | ||
| import com.shorrockin.cascal.session.Insert | ||
| import com.shorrockin.cascal.session.RangePredicate | ||
| import com.shorrockin.cascal.session.Order | ||
|
|
||
| class CompositeTest extends EmbeddedCassandra { | ||
|
|
||
| @Test def composite2InsertGet = borrow { session => | ||
| val name = ("composite name", 1) | ||
| val col = "Test" \ "Composite2" \ "Insert Get" \ name | ||
| session.insert(col \ "composite value") | ||
|
|
||
| val colName = session.get(col).get.name | ||
| assertEquals(name, tuple[String, Int](session.get(col).get.name)) | ||
| } | ||
|
|
||
| @Test def composite3InsertGet = borrow { session => | ||
| val name = ("composite name", 1L, "name part 3") | ||
| val col = "Test" \ "Composite3" \ "Insert Get" \ name | ||
| session.insert(col \ "composite value") | ||
|
|
||
| val colName = session.get(col).get.name | ||
| assertEquals(name, tuple[String, Long, String](session.get(col).get.name)) | ||
| } | ||
|
|
||
| @Test def composite2Range = borrow { session => | ||
| val key = "Test" \ "Composite2" \ "Composite Range" | ||
| val col1 = key \ (("composite", 1)) \ 1 | ||
| val col2 = key \ (("composite", 2)) \ 1 | ||
| val col3 = key \ (("composite", 3)) \ 1 | ||
| val col4 = key \ (("composite", 4)) \ 1 | ||
| val col5 = key \ (("composite", 5)) \ 1 | ||
| val col6 = key \ (("comcom", 5)) \ 1 | ||
|
|
||
| session.batch(Insert(col1) :: Insert(col2) :: Insert(col3) :: Insert(col4) :: Insert(col5) :: Insert(col6)) | ||
|
|
||
| val result1 = session.list(key, RangePredicate(Some(("c", None)), None, Order.Ascending, None)) | ||
| assertEquals(6, result1.size) | ||
|
|
||
| val result2 = session.list(key, RangePredicate(Some(("composite", None)), None, Order.Ascending, None)) | ||
| assertEquals(5, result2.size) | ||
|
|
||
| val result3 = session.list(key, RangePredicate(("composite", 2), ("composite", 4))) | ||
| assertEquals(3, result3.size) | ||
| assertEquals(col2, result3(0)) | ||
| assertEquals(col3, result3(1)) | ||
| assertEquals(col4, result3(2)) | ||
| } | ||
|
|
||
| @Test def composite3Range = borrow { session => | ||
| val key = "Test" \ "Composite3" \ "Composite Range" | ||
| val col1 = key \ (("composite", 1L, "A")) \ 1 | ||
| val col2 = key \ (("composite", 1L, "a")) \ 1 | ||
| val col3 = key \ (("composite", 10L, "A")) \ 1 | ||
|
|
||
| session.batch(Insert(col1) :: Insert(col2) :: Insert(col3)) | ||
|
|
||
| val result0 = session.list(key, RangePredicate(Some(("composite", None)), None, Order.Descending, None)) | ||
| assertEquals(0, result0.size) | ||
|
|
||
| val result1 = session.list(key, RangePredicate(Some(("composite", None)), None, Order.Ascending, None)) | ||
| assertEquals(3, result1.size) | ||
|
|
||
| val result2 = session.list(key, RangePredicate(("composite", 1L), ("composite", 2L))) | ||
| assertEquals(2, result2.size) | ||
| assertEquals(col1, result2(0)) | ||
| assertEquals(col2, result2(1)) | ||
|
|
||
| val result3 = session.list(key, RangePredicate(("composite", 1L, "B"), ("composite", 2L))) | ||
| assertEquals(1, result3.size) | ||
| assertEquals(col2, result3(0)) | ||
| } | ||
| } |