How to develop an efficient binary file protocol with Scodec and Akka Streams

build.sbt

libraryDependencies ++= List(
  "org.scodec" %% "scodec-core" % "1.9.0",
  "org.scodec" %% "scodec-bits" % "1.1.0"
)

demo.scala

import scodec._
import bits._
import codecs._

scodec.codecs.int32
// res0: scodec.Codec[Int] = 32-bit signed integer

int32 encode 42
// res1: scodec.Attempt[scodec.bits.BitVector] = Successful(BitVector(32 bits, 0x0000002a))

// getting the binary representation
(int32 encode 42).getOrElse(BitVector.empty).toBin
// res3: String = 00000000000000000000000000101010

// define a codec for an 8-bit unsigned int followed by an 8-bit unsigned int followed by a 16-bit unsigned int.
val codec = (uint8 ~ uint8 ~ uint16)
// codec: scodec.Codec[((Int, Int), Int)] = ((8-bit unsigned integer, 8-bit unsigned integer), 16-bit unsigned integer)

// using this codec to decode from a from bits
codec.decode(hex"0x2a2a0539".bits)
// res6: scodec.Attempt[scodec.DecodeResult[((Int, Int), Int)]] = Successful(DecodeResult(((42,42),1337),BitVector(empty)))

// using shapeless compile time magic for case class codecs
case class Point(x: Int, y: Int)
// defined class Point

val pointCodec = (int32 :: int32).as[Point]
// pointCodec: scodec.Codec[Point] = scodec.Codec$$anon$2@1b9cc190

case class Tick(time: Long, bid: Double, ask: Double)

utf8 encode "1420148801108,1.20989,1.21049"
// res8: scodec.Attempt[scodec.bits.BitVector] = Successful(BitVector(232 bits, 0x313432303134383830313130382c312e32303938392c312e3231303439))

val tickCodec = (long(64) :: double :: double).as[Tick]
// tickCodec: scodec.Codec[Tick] = scodec.Codec$$anon$2@4456705

tickCodec encode Tick(1420148801108L, 1.20989, 1.21049)
// res9: scodec.Attempt[scodec.bits.BitVector] = Successful(BitVector(192 bits, 0x0000014aa776fe543ff35bb59ddc1e793ff35e2ac3222920))

val tick = Tick(1420148801108L, 1.20989, 1.21049)
// tick: Tick = Tick(1420148801108,1.20989,1.21049)

case class FactorizedTick(time: Long, bid: Int, ask: Int)
// defined class FactorizedTick

val factorizedTickCodec = (long(64) :: int32 :: int32).as[FactorizedTick]
// factorizedTickCodec: scodec.Codec[FactorizedTick] = scodec.Codec$$anon$2@2bff5f88

implicit final class TickOps(private val wrappedTick: Tick) extends AnyVal {
  def factorize: FactorizedTick =
    FactorizedTick(
      wrappedTick.time,
      (wrappedTick.bid * 100000).toInt,
      (wrappedTick.ask * 100000).toInt
    )
}
// defined class TickOps

factorizedTickCodec encode tick.factorize
// res10: scodec.Attempt[scodec.bits.BitVector] = Successful(BitVector(128 bits, 0x0000014aa776fe540001d89c0001d8d9))

case class FactorizedDeltaTick(timeDelta: Long, bidDelta: Int, askDelta: Int)
// defined class FactorizedDeltaTick

val factorizedDeltaTickCodec = (long(64) :: int32 :: int32).as[FactorizedDeltaTick]
// factorizedDeltaTickCodec: scodec.Codec[FactorizedDeltaTick] = scodec.Codec$$anon$2@4243eeb5

implicit final class FactorizedTickOps(private val wrappedFactorizedTick: FactorizedTick) extends AnyVal {
  def deltaTo(prevFactorizedTick: FactorizedTick): FactorizedDeltaTick =
    FactorizedDeltaTick(
      wrappedFactorizedTick.time - prevFactorizedTick.time,
      wrappedFactorizedTick.bid - prevFactorizedTick.bid,
      wrappedFactorizedTick.ask - prevFactorizedTick.ask
    )
}
// defined class FactorizedTickOps

val otherTick = Tick(1420148801207L, 1.21004, 1.21063)
// otherTick: Tick = Tick(1420148801207,1.21004,1.21063)

val delta = otherTick.factorize deltaTo tick.factorize
// delta: FactorizedDeltaTick = FactorizedDeltaTick(99,16,14)

factorizedDeltaTickCodec encode delta
// res11: scodec.Attempt[scodec.bits.BitVector] = Successful(BitVector(128 bits, 0x0000000000000063000000100000000e))

// saving 75%
int32 encode 42
// res13: scodec.Attempt[scodec.bits.BitVector] = Successful(BitVector(32 bits, 0x0000002a))

vint encode 42
// res14: scodec.Attempt[scodec.bits.BitVector] = Successful(BitVector(8 bits, 0x2a))

// saving 50%
int32 encode 1337
// res16: scodec.Attempt[scodec.bits.BitVector] = Successful(BitVector(32 bits, 0x00000539))

vint encode 1337
// res17: scodec.Attempt[scodec.bits.BitVector] = Successful(BitVector(16 bits, 0xb90a))

// using the same space
int32 encode 134217728
// res19: scodec.Attempt[scodec.bits.BitVector] = Successful(BitVector(32 bits, 0x08000000))

vint encode 134217728
// res20: scodec.Attempt[scodec.bits.BitVector] = Successful(BitVector(32 bits, 0x80808040))

// using 25% more space
int32 encode 2147483647
// res22: scodec.Attempt[scodec.bits.BitVector] = Successful(BitVector(32 bits, 0x7fffffff))

vint encode 2147483647
// res23: scodec.Attempt[scodec.bits.BitVector] = Successful(BitVector(40 bits, 0xffffffff07))

val factorizedDeltaTickCodecV = (vlong :: vint :: vint).as[FactorizedDeltaTick]
// factorizedDeltaTickCodecV: scodec.Codec[FactorizedDeltaTick] = scodec.Codec$$anon$2@781b8436

factorizedDeltaTickCodecV encode delta
// res24: scodec.Attempt[scodec.bits.BitVector] = Successful(BitVector(24 bits, 0x63100e))

vint encode -5
// res25: scodec.Attempt[scodec.bits.BitVector] = Successful(BitVector(40 bits, 0xfbffffff0f))

val negativeDelta = FactorizedDeltaTick(10, -5, -8)
// negativeDelta: FactorizedDeltaTick = FactorizedDeltaTick(10,-5,-8)

factorizedDeltaTickCodec encode negativeDelta
// res26: scodec.Attempt[scodec.bits.BitVector] = Successful(BitVector(128 bits, 0x000000000000000afffffffbfffffff8))

case class NonNegativeFactorizedDeltaTick(timeDelta: Long, bidDeltaNeg: Boolean, bidDelta: Int, askDeltaNeg: Boolean, askDelta: Int)
// defined class NonNegativeFactorizedDeltaTick

val nonNegFactorizedDeltaTickCodecV = (vlong :: bool :: vint :: bool :: vint).as[NonNegativeFactorizedDeltaTick]
// nonNegFactorizedDeltaTickCodecV: scodec.Codec[NonNegativeFactorizedDeltaTick] = scodec.Codec$$anon$2@77269d0a

implicit class FactorizedDeltaTickOps(private val wrappedFactorizedDeltaTick: FactorizedDeltaTick) extends AnyVal {

  def nonNegative: NonNegativeFactorizedDeltaTick = NonNegativeFactorizedDeltaTick(
    wrappedFactorizedDeltaTick.timeDelta,
    wrappedFactorizedDeltaTick.bidDelta < 0,
    wrappedFactorizedDeltaTick.bidDelta.abs,
    wrappedFactorizedDeltaTick.askDelta < 0,
    wrappedFactorizedDeltaTick.askDelta.abs
  )
}
// defined class FactorizedDeltaTickOps

val nonNegativeDelta = negativeDelta.nonNegative
// nonNegativeDelta: NonNegativeFactorizedDeltaTick = NonNegativeFactorizedDeltaTick(10,true,5,true,8)

nonNegFactorizedDeltaTickCodecV encode nonNegativeDelta
// res27: scodec.Attempt[scodec.bits.BitVector] = Successful(BitVector(26 bits, 0x0a82c20))