From 43c0bce39ba701dd1d8352ea9b10130cf88ce977 Mon Sep 17 00:00:00 2001
From: Michael Burman <yak@iki.fi>
Date: Mon, 5 Sep 2016 17:08:52 +0300
Subject: [PATCH] Fix empty block case

---
 .../ts/compression/gorilla/Decompressor.java  |  5 +++++
 .../ts/compression/gorilla/EncodeTest.java    | 19 +++++++++++++++++++
 2 files changed, 24 insertions(+)

diff --git a/src/main/java/fi/iki/yak/ts/compression/gorilla/Decompressor.java b/src/main/java/fi/iki/yak/ts/compression/gorilla/Decompressor.java
index ea7080d..9ea50bb 100644
--- a/src/main/java/fi/iki/yak/ts/compression/gorilla/Decompressor.java
+++ b/src/main/java/fi/iki/yak/ts/compression/gorilla/Decompressor.java
@@ -36,6 +36,11 @@ public Pair readPair() {
         if (storedTimestamp == 0) {
             // First item to read
             storedDelta = in.getLong(Compressor.FIRST_DELTA_BITS);
+
+            if(storedDelta == (1<<27) - 1) {
+                // Empty - no timestamp space left
+                return null;
+            }
             storedVal = Double.longBitsToDouble(in.getLong(64));
             storedTimestamp = blockTimestamp + storedDelta;
         } else {
diff --git a/src/test/java/fi/iki/yak/ts/compression/gorilla/EncodeTest.java b/src/test/java/fi/iki/yak/ts/compression/gorilla/EncodeTest.java
index f46b9c4..cc38f01 100644
--- a/src/test/java/fi/iki/yak/ts/compression/gorilla/EncodeTest.java
+++ b/src/test/java/fi/iki/yak/ts/compression/gorilla/EncodeTest.java
@@ -142,4 +142,23 @@ void testEncodeLargeAmountOfData() throws Exception {
         }
         assertNull(d.readPair());
     }
+
+    @Test
+    void testEmptyBlock() throws Exception {
+        long now = LocalDateTime.now().truncatedTo(ChronoUnit.HOURS)
+                .toInstant(ZoneOffset.UTC).toEpochMilli();
+
+        ByteBufferBitOutput output = new ByteBufferBitOutput();
+
+        Compressor c = new Compressor(now, output);
+        c.close();
+
+        ByteBuffer byteBuffer = output.getByteBuffer();
+        byteBuffer.flip();
+
+        ByteBufferBitInput input = new ByteBufferBitInput(byteBuffer);
+        Decompressor d = new Decompressor(input);
+
+        assertNull(d.readPair());
+    }
 }