Repository: liftbridge-io/liftbridge (go)

Issue:
Server does not correctly handle empty message values
The server does not properly handle null/empty message values. The following panic happens when trying to subscribe to a stream after publishing some messages with empty values:

```
DEBU[2019-12-24 14:11:18] api: Subscribe [stream=xxxxxxxxxx, partition=0, start=EARLIEST, offset=0, timestamp=-6795364578871345152]
panic: runtime error: slice bounds out of range

goroutine 118 [running]:
github.com/liftbridge-io/liftbridge/server/commitlog.Message.Headers(0xc0000a06c0, 0x32, 0x32, 0xc0000e49e0)
        /Users/tylertreat/Go/src/github.com/liftbridge-io/liftbridge/server/commitlog/message.go:53 +0x302
github.com/liftbridge-io/liftbridge/server.(*apiServer).subscribe.func1()
        /Users/tylertreat/Go/src/github.com/liftbridge-io/liftbridge/server/api.go:258 +0x156
github.com/liftbridge-io/liftbridge/server.(*Server).startGoroutine.func1(0xc0001bb2c0, 0xc0001c2820)
        /Users/tylertreat/Go/src/github.com/liftbridge-io/liftbridge/server/server.go:768 +0x27
created by github.com/liftbridge-io/liftbridge/server.(*Server).startGoroutine
        /Users/tylertreat/Go/src/github.com/liftbridge-io/liftbridge/server/server.go:767 +0x97
```

Pull request:
Handle messages with nil values
Fixes #147.

Patch:
diff --git a/server/commitlog/message.go b/server/commitlog/message.go
index 6579560..147cd36 100644
--- a/server/commitlog/message.go
+++ b/server/commitlog/message.go
@@ -75,6 +75,9 @@ func (m Message) valueOffsets() (start, end, size int32) {
 	_, keyEnd, _ := m.keyOffsets()
 	start = keyEnd
 	size = int32(proto.Encoding.Uint32(m[start:]))
-	end = start + 4 + size
+	end = start + 4
+	if size != -1 {
+		end += size
+	}
 	return
 }


Test patch:
diff --git a/server/commitlog/commitlog_test.go b/server/commitlog/commitlog_test.go
index d09cee7..d0e2d26 100644
--- a/server/commitlog/commitlog_test.go
+++ b/server/commitlog/commitlog_test.go
@@ -22,6 +22,7 @@ var (
 		{Value: []byte("two"), Timestamp: 2, LeaderEpoch: 42, Headers: headers},
 		{Value: []byte("three"), Timestamp: 3, LeaderEpoch: 42, Headers: headers},
 		{Value: []byte("four"), Timestamp: 4, LeaderEpoch: 42, Headers: headers},
+		{Value: nil, Timestamp: 5, LeaderEpoch: 42, Headers: headers},
 	}
 )
 
@@ -66,7 +67,7 @@ func TestAppendMessageSet(t *testing.T) {
 
 	offsets, err := l.AppendMessageSet(set)
 	require.NoError(t, err)
-	require.Equal(t, []int64{0, 1, 2, 3}, offsets)
+	require.Equal(t, []int64{0, 1, 2, 3, 4}, offsets)
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 	r, err := l.NewReader(0, true)