bitpeace – Handle binary data with doobie

This is a library to store and load binary data based on doobie.

It stores binary files in chunks and offers ways to retrieve parts of each file, making it for example useful when serving range requests over http.

Using

Bitpeace is available from maven central for scala 2.12 and from version 0.4 for scala 2.13.

"com.github.eikek" %% "bitpeace-core" % "0.7.0"

Dependencies

It obviously depends on doobie and therefore on cats and fs2.

While trying to minimize further dependencies, I choose to include two more:

• tika-core (optional), Apache 2.0 License, http://tika.apache.org
• scodec-bits, BSD-3-Clause, https://github.com/scodec/scodec-bits

Both dependencies are itself “dependency-free”. The tika-core is marked as optional.

Since mimetype detection belongs to “handling binaries”, I wanted to include it in this library. Tika is used for mimetype detection by default, but can be replaced by custom code. Therefore it is added as optional dependency and you need to declare it in case you want to use it.

The scodec-bits library is used for its ByteVector class; to avoid usages of Array[Byte].

DBMS Support

DML statements (Queries and insert/update) are standard SQL and should work on most database systems.

The code to create the schema BitpeaceTables works with Postgres and H2. For other database systems, simply create the two tables yourself.

Setup

Bitpeace needs a small config that defines the table names to use and two functions:

• instance of MimetypeDetect to detect mimetype given some bytes
• generate a random string to use as an primary key

There is a default config:

import _root_.bitpeace._, cats.effect.IO

val cfg1 = BitpeaceConfig.default[IO]
// cfg1: BitpeaceConfig[IO] = BitpeaceConfig(
//   metaTable = "FileMeta",
//   chunkTable = "FileChunk",
//   mimetypeDetect = bitpeace.MimetypeDetect$$anon$1@62bf85d8,
//   randomIdGen = Delay(
//     thunk = bitpeace.BitpeaceConfig$$$Lambda$8156/1667373404@7f35e600
//   )
// )

It uses javas UUID class to generate random ids and has no ability to detect mimetypes. The mimetype will always be application/octet-stream for all files.

If you add tika-core to your project, you can use the other default:

val cfg2 = BitpeaceConfig.defaultTika[IO]
// cfg2: BitpeaceConfig[IO] = BitpeaceConfig(
//   metaTable = "FileMeta",
//   chunkTable = "FileChunk",
//   mimetypeDetect = bitpeace.TikaMimetypeDetect$@33289f16,
//   randomIdGen = Delay(
//     thunk = bitpeace.BitpeaceConfig$$$Lambda$8156/1667373404@7f35e600
//   )
// )

which only differs in that the MimetypeDetect is now implemented using the tika library.

The second requirement is a doobie Transactor to connect to the database. For example, this creates one for the H2 database:

import doobie._, doobie.implicits._

implicit val CS = IO.contextShift(scala.concurrent.ExecutionContext.global)
// CS: cats.effect.ContextShift[IO] = cats.effect.internals.IOContextShift@56ae5d02

val xa = Transactor.fromDriverManager[IO](
  "org.h2.Driver", s"jdbc:h2:/tmp/bitpeace-testdb", "sa", ""
)
// xa: Transactor.Aux[IO, Unit] = doobie.util.transactor$Transactor$$anon$13@536ac475

Given a config and a transactor, the main entrypoint Bitpeace can be created:

val bitpeace = Bitpeace(BitpeaceConfig.defaultTika[IO], xa)
// bitpeace: Bitpeace[IO[A]] = bitpeace.Bitpeace$$anon$1@507ad76c

In order to start using it, the database schema must exist. The BitpeaceTables class is a convenience helper to do that:

BitpeaceTables(BitpeaceConfig.default[IO]).create(sql.Dbms.H2).transact(xa).unsafeRunSync

Usage

Storing data

The data to store is given as a Stream[F, Byte]. A chunksize must be specified that defines how many bytes are stored in one blob object. Other two parameters involve a hint to support mimetype detection (for example the filename) and a timestamp associated to that file.

saveNew

Data can be inserted using saveNew:

import fs2._
import scodec.bits.ByteVector

val chunksize = 128 * 1024
// chunksize: Int = 131072
val data = Stream.chunk[IO, Byte](Chunk.byteVector(ByteVector.fromValidHex("68656c6c6f20776f726c64")))
// data: Stream[IO, Byte] = Stream(..)
val meta = bitpeace.saveNew(data, chunksize, MimetypeHint.none)
// meta: Stream[IO[A], FileMeta] = Stream(..)
val savedFileMeta = meta.compile.lastOrError.unsafeRunSync
// savedFileMeta: FileMeta = FileMeta(
//   id = "28b4ea25-529b-48be-998f-ced007cfca9f",
//   timestamp = 2021-03-10T21:48:12.940Z,
//   mimetype = Mimetype(primary = "text", sub = "plain", params = Map()),
//   length = 11L,
//   checksum = "b94d27b9934d3e08a52e52d7da7dabfac484efe37a5380ee9088f7ace2efcde9",
//   chunks = 1,
//   chunksize = 131072
// )

The FileMeta return value contains some meta data about the input data, like a sha checksum, size, chunksize and the mimetype. Its id can later be used to get the data back out.

The id is generated using the random id generation function from the config. You can supply a custom fixed id for a file, too.

bitpeace.saveNew(data, chunksize, MimetypeHint.none, fileId = Some("abc123"))
// res1: Stream[IO[A], FileMeta] = Stream(..)

save (no duplicates)

The saveNew command simply inserts the data generating a random id. If you don't want duplicates, you can run makeUnique:

val out = bitpeace.makeUnique(savedFileMeta)
// out: Stream[IO[A], Outcome[FileMeta]] = Stream(..)

This will check if there is a file with the same checksum. If true, the given FileMeta (and data) is deleted and the just found value is returned. This is encoded in the Outcome.Unmodified value.

If there is no such file, the id of the given FileMeta is updated to be its checksum (to ensure no duplicates). Then it is returned inside a Outcome.Created indicating that the given data has been used. If during the id update an error occurs, it may be caused by storing the same file concurrently. Then it is again tried to lookup an existing file.

You can combine those two operations:

bitpeace.saveNew(data, chunksize, MimetypeHint.none).flatMap(bitpeace.makeUnique)
// res2: Stream[IO[x], Outcome[FileMeta]] = Stream(..)

or use the operation save (which is just a shortcut for the above):

bitpeace.save(data, chunksize, MimetypeHint.none)
// res3: Stream[IO[A], Outcome[FileMeta]] = Stream(..)

addChunks

The third case is when chunks of data arrive in some random order. Then you can use addChunk:

val chunk = FileChunk("file-id", 1, ByteVector.fromValidHex("68656c6c6f20776f726c64"))
// chunk: FileChunk = FileChunk(
//   fileId = "file-id",
//   chunkNr = 1L,
//   chunkData = Chunk(
//     bytes = View(
//       at = scodec.bits.ByteVector$AtArray@61a641ee,
//       offset = 0L,
//       size = 11L
//     )
//   )
// )
bitpeace.addChunk(chunk, chunksize, 12, MimetypeHint.none)
// res4: Stream[IO[A], Outcome[FileMeta]] = Stream(..)

It is necessary to tell when the last chunk arrives, to calculate the checksum and set the timestamp. That's why you either need to tell the total number of chunks (it is the 12 above), or the total length of the file together with the intended chunksize.

The operation returns the updated FileMeta object and you can tell whether the data is complete if the length and checksum are set. The result is wrapped in a Outcome to tell whether the chunk already existed or not.

Chunks must be 0-indexed!

Retrieving data

The id to identify the FileMeta object is required to retrieve data. With a FileMeta object, one can stream the bytes using either fetchData or fetchData2.

val id: String = "xyz123"
// id: String = "xyz123"
val meta2 = bitpeace.get(id)
// meta2: Stream[IO[A], Option[FileMeta]] = Stream(..)
val data2 = meta.through(bitpeace.fetchData(RangeDef.all))
// data2: Stream[IO[x], Byte] = Stream(..)

The difference between fetchData and fetchData2 is that the former uses one connection per chunk, whereas the latter uses one connection for the entire stream (i.e. it is closed once the stream terminates).

The fetchData operations expect a RangeDef argument. This can be used to return a specific byte range. A RangeDef is a function from FileMeta and a range request to a Range. Since range requests can be wrong (i.e. exceed total length), the return is wrapped in a cats.data.Validated. The RangeDef companion object contains several methods to construct RangeDefs. For example:

// get the first chunk only
bitpeace.fetchData(RangeDef.firstChunk)
// res5: Stream[IO[A], FileMeta] => Stream[IO[A], Byte] = bitpeace.Bitpeace$$anon$1$$Lambda$8512/1196453114@135739b9

// get the first x bytes
bitpeace.fetchData(RangeDef.firstBytes(1024))
// res6: Stream[IO[A], FileMeta] => Stream[IO[A], Byte] = bitpeace.Bitpeace$$anon$1$$Lambda$8512/1196453114@173f1aed

// get next 2K bytes skipping 4K bytes
bitpeace.fetchData(RangeDef.bytes(Some(4 * 1024), Some(2 * 1024)))
// res7: Stream[IO[A], FileMeta] => Stream[IO[A], Byte] = bitpeace.Bitpeace$$anon$1$$Lambda$8512/1196453114@b4300bd

// get all remaining bytes after skipping 4K
bitpeace.fetchData(RangeDef.bytes(Some(4 * 1024), None))
// res8: Stream[IO[A], FileMeta] => Stream[IO[A], Byte] = bitpeace.Bitpeace$$anon$1$$Lambda$8512/1196453114@5839890a

Misc

The library is distributed using the MIT license.

Feedback is very welcome! Put it in a mail to eikek at posteo.de or the issue tracker.