do not hang request on entity not found (#78)

datastreamer/datastreamer_test.go

@@ -72,6 +72,10 @@ var (
 	testBookmark = TestBookmark{
 		FieldA: []byte{0, 1, 0, 0, 0, 0, 0, 0, 0},
 	}
+
+	nonAddedBookmark = TestBookmark{
+		FieldA: []byte{0, 2, 0, 0, 0, 0, 0, 0, 0},
+	}
 )
 
 func deleteFiles() error {
@@ -146,11 +150,22 @@ func TestClient(t *testing.T) {
 	err = client.Start()
 	require.NoError(t, err)
 
+	// Should succeed
 	client.FromBookmark = testBookmark.Encode()
 	err = client.ExecCommand(datastreamer.CmdBookmark)
 	require.NoError(t, err)
 	require.Equal(t, testEntry.Encode(), client.Entry.Data)
 
+	// Should fail because the bookmark is not added
+	client.FromBookmark = nonAddedBookmark.Encode()
+	err = client.ExecCommand(datastreamer.CmdBookmark)
+	require.EqualError(t, datastreamer.ErrEntryNotFound, err.Error())
+
+	// Should fail because the entry is not added
+	client.FromEntry = 3
+	err = client.ExecCommand(datastreamer.CmdEntry)
+	require.EqualError(t, datastreamer.ErrEntryNotFound, err.Error())
+
 	client.FromEntry = 2
 	err = client.ExecCommand(datastreamer.CmdEntry)
 	require.NoError(t, err)

datastreamer/streamclient.go

@@ -171,9 +171,15 @@ func (c *StreamClient) ExecCommand(cmd Command) error {
 		c.Header = h
 	case CmdEntry:
 		e := c.getEntry()
+		if e.Type == EntryTypeNotFound {
+			return ErrEntryNotFound
+		}
 		c.Entry = e
 	case CmdBookmark:
 		e := c.getEntry()
+		if e.Type == EntryTypeNotFound {
+			return ErrEntryNotFound
+		}
 		c.Entry = e
 	}
 

datastreamer/streamserver.go

@@ -3,6 +3,7 @@ package datastreamer
 import (
 	"encoding/binary"
 	"io"
+	"math"
 	"net"
 	"strconv"
 	"strings"
@@ -31,6 +32,9 @@ type StreamType uint64
 // CommandError type for the command responses
 type CommandError uint32
 
+// EntryTypeNotFound is used to indicate that the entry type is not found
+const EntryTypeNotFound = math.MaxUint32
+
 const (
 	maxConnections = 100 // Maximum number of connected clients
 	streamBuffer   = 256 // Buffers for the stream channel
@@ -823,7 +827,10 @@ func (s *StreamServer) processCmdEntry(clientId string) error {
 	// Get the requested entry
 	entry, err := s.GetEntry(entryNumber)
 	if err != nil {
-		return err
+		log.Errorf("Error getting entry %d: %v", entryNumber, err)
+		entry = FileEntry{}
+		entry.Length = FixedSizeFileEntry
+		entry.Type = EntryTypeNotFound
 	}
 	entry.packetType = PtDataRsp
 	binaryEntry := encodeFileEntryToBinary(entry)
@@ -870,7 +877,10 @@ func (s *StreamServer) processCmdBookmark(clientId string) error {
 	// Get the requested bookmark
 	entry, err := s.GetFirstEventAfterBookmark(bookmark)
 	if err != nil {
-		return err
+		log.Errorf("Error getting bookmark %v: %v", bookmark, err)
+		entry = FileEntry{}
+		entry.Length = FixedSizeFileEntry
+		entry.Type = EntryTypeNotFound
 	}
 	entry.packetType = PtDataRsp
 	binaryEntry := encodeFileEntryToBinary(entry)