Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
util-core: Move chunked and framed to BufReader
Problem/Solution: Reader.chunked and Reader.framed are Buf specific, move them to BufReader. JIRA Issues: CSL-6756 Differential Revision: https://phabricator.twitter.biz/D392198
- Loading branch information
Showing
5 changed files
with
188 additions
and
190 deletions.
There are no files selected for viewing
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
139 changes: 109 additions & 30 deletions
139
util-core/src/test/scala/com/twitter/io/BufReaderTest.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 |
|---|---|---|
| @@ -1,69 +1,148 @@ | ||
| package com.twitter.io | ||
|
|
||
| import com.twitter.conversions.DurationOps._ | ||
| import com.twitter.conversions.StorageUnitOps._ | ||
| import com.twitter.util.{Await, Future} | ||
| import org.scalacheck.Prop.forAll | ||
| import org.scalacheck.Gen | ||
| import org.scalacheck.{Arbitrary, Gen} | ||
| import org.scalatest.FunSuite | ||
| import org.scalatestplus.scalacheck.Checkers | ||
| import org.scalatestplus.scalacheck.ScalaCheckDrivenPropertyChecks | ||
| import scala.annotation.tailrec | ||
|
|
||
| class BufReaderTest extends FunSuite with Checkers { | ||
| class BufReaderTest extends FunSuite with ScalaCheckDrivenPropertyChecks { | ||
|
|
||
| private def await[A](f: Future[A]): A = Await.result(f, 5.seconds) | ||
|
|
||
| test("BufReader") { | ||
| check { bytes: String => | ||
| forAll { bytes: String => | ||
| val buf = Buf.Utf8(bytes) | ||
| val r = Reader.fromBuf(buf, 8) | ||
| await(BufReader.readAll(r)) == buf | ||
| assert(await(BufReader.readAll(r)) == buf) | ||
| } | ||
| } | ||
|
|
||
| test("BufReader - discard") { | ||
| val alphaString = for { | ||
| val stringAndChunk = for { | ||
| s <- Gen.alphaStr | ||
| } yield s | ||
| val positiveInt = for { | ||
| n <- Gen.posNum[Int] | ||
| } yield n | ||
| i <- Gen.posNum[Int].suchThat(_ <= s.length) | ||
| } yield (s, i) | ||
|
|
||
| check { | ||
| forAll(alphaString, positiveInt) { (bytes, n) => | ||
| forAll(stringAndChunk) { | ||
| case (bytes, n) => | ||
| val r = Reader.fromBuf(Buf.Utf8(bytes), n) | ||
| r.discard() | ||
|
|
||
| bytes.length == 0 || | ||
| Await | ||
| .ready(r.read(), 5.seconds).poll.exists( | ||
| _.throwable.isInstanceOf[ReaderDiscardedException] | ||
| ) | ||
| } | ||
| assert( | ||
| bytes.length == 0 || | ||
| Await | ||
| .ready(r.read(), 5.seconds).poll | ||
| .exists(_.throwable.isInstanceOf[ReaderDiscardedException])) | ||
| } | ||
| } | ||
|
|
||
| test("BufReader - iterator") { | ||
| val alphaString = for { | ||
| val stringAndChunk = for { | ||
| s <- Gen.alphaStr | ||
| } yield s | ||
| val positiveInt = for { | ||
| n <- Gen.posNum[Int] | ||
| } yield n | ||
| i <- Gen.posNum[Int].suchThat(_ <= s.length) | ||
| } yield (s, i) | ||
|
|
||
| check { | ||
| forAll(alphaString, positiveInt) { (bytes, n) => | ||
| forAll(stringAndChunk) { | ||
| case (bytes, n) => | ||
| var result = Buf.Empty | ||
| val iterator = BufReader.iterator(Buf.Utf8(bytes), n) | ||
| while (iterator.hasNext) { | ||
| result = result.concat(iterator.next) | ||
| } | ||
| Buf.Utf8(bytes) == result | ||
| } | ||
| assert(Buf.Utf8(bytes) == result) | ||
| } | ||
| } | ||
|
|
||
| test("BufReader - readAll") { | ||
| check { bytes: String => | ||
| forAll { bytes: String => | ||
| val r = Reader.fromBuf(Buf.Utf8(bytes)) | ||
| await(BufReader.readAll(r)) == Buf.Utf8(bytes) | ||
| assert(await(BufReader.readAll(r)) == Buf.Utf8(bytes)) | ||
| } | ||
| } | ||
|
|
||
| test("BufReader.chunked") { | ||
| val stringAndChunk = for { | ||
| s <- Gen.alphaStr | ||
| i <- Gen.posNum[Int].suchThat(_ <= s.length) | ||
| } yield (s, i) | ||
|
|
||
| forAll(stringAndChunk) { | ||
| case (s, i) => | ||
| val r = BufReader.chunked(Reader.fromBuf(Buf.Utf8(s), 32), i) | ||
|
|
||
| def readLoop(): Unit = await(r.read()) match { | ||
| case Some(b) => | ||
| assert(b.length <= i) | ||
| readLoop() | ||
| case None => () | ||
| } | ||
|
|
||
| readLoop() | ||
| } | ||
| } | ||
|
|
||
| test("BufReader.framed reads framed data") { | ||
| val getByteArrays: Gen[Seq[Buf]] = Gen.listOf( | ||
| for { | ||
| // limit arrays to a few kilobytes, otherwise we may generate a very large amount of data | ||
| numBytes <- Gen.choose(0.bytes.inBytes, 2.kilobytes.inBytes) | ||
| bytes <- Gen.containerOfN[Array, Byte](numBytes.toInt, Arbitrary.arbitrary[Byte]) | ||
| } yield Buf.ByteArray.Owned(bytes) | ||
| ) | ||
|
|
||
| forAll(getByteArrays) { buffers: Seq[Buf] => | ||
| val buffersWithLength = buffers.map(buf => Buf.U32BE(buf.length).concat(buf)) | ||
|
|
||
| val r = BufReader.framed(BufReader(Buf(buffersWithLength)), new BufReaderTest.U32BEFramer()) | ||
|
|
||
| // read all of the frames | ||
| buffers.foreach { buf => | ||
| assert(await(r.read()).contains(buf)) | ||
| } | ||
|
|
||
| // make sure the reader signals EOF | ||
| assert(await(r.read()).isEmpty) | ||
| } | ||
| } | ||
|
|
||
| test("BufReader.framed reads empty frames") { | ||
| val r = BufReader.framed(BufReader(Buf.U32BE(0)), new BufReaderTest.U32BEFramer()) | ||
| assert(await(r.read()).contains(Buf.Empty)) | ||
| assert(await(r.read()).isEmpty) | ||
| } | ||
|
|
||
| test("Framed works on non-list collections") { | ||
| val r = BufReader.framed(BufReader(Buf.U32BE(0)), i => Vector(i)) | ||
| assert(await(r.read()).isDefined) | ||
| } | ||
| } | ||
|
|
||
| object BufReaderTest { | ||
|
|
||
| /** | ||
| * Used to test BufReader.framed, extract fields in terms of | ||
| * frames, signified by a 32-bit BE value preceding | ||
| * each frame. | ||
| */ | ||
| private class U32BEFramer() extends (Buf => Seq[Buf]) { | ||
| var state: Buf = Buf.Empty | ||
|
|
||
| @tailrec | ||
| private def loop(acc: Seq[Buf], buf: Buf): Seq[Buf] = { | ||
| buf match { | ||
| case Buf.U32BE(l, d) if d.length >= l => | ||
| loop(acc :+ d.slice(0, l), d.slice(l, d.length)) | ||
| case _ => | ||
| state = buf | ||
| acc | ||
| } | ||
| } | ||
|
|
||
| def apply(buf: Buf): Seq[Buf] = synchronized { | ||
| loop(Seq.empty, state concat buf) | ||
| } | ||
| } | ||
| } |
Oops, something went wrong.