Remove FragmentHandler's error response

aeron/fragmentassembler.go

@@ -67,10 +67,10 @@ func (f *FragmentAssembler) OnFragment(
 	buffer *atomic.Buffer,
 	offset int32,
 	length int32,
-	header *logbuffer.Header) error {
+	header *logbuffer.Header) {
 	flags := header.Flags()
 	if (flags & unfragmented) == unfragmented {
-		return f.delegate(buffer, offset, length, header)
+		f.delegate(buffer, offset, length, header)
 	} else {
 		if (flags & beginFrag) == beginFrag {
 			builder, ok := f.builderBySessionIdMap[header.SessionId()]
@@ -86,7 +86,7 @@ func (f *FragmentAssembler) OnFragment(
 				buffer.WriteBytes(builder, offset, length)
 				if (flags & endFrag) == endFrag {
 					msgLength := builder.Len()
-					return f.delegate(
+					f.delegate(
 						atomic.MakeBuffer(builder.Bytes(), msgLength),
 						int32(0),
 						int32(msgLength),
@@ -96,5 +96,4 @@ func (f *FragmentAssembler) OnFragment(
 			}
 		}
 	}
-	return nil
 }

aeron/image.go

@@ -21,9 +21,9 @@ import "github.com/lirm/aeron-go/aeron/logbuffer/term"
 // the many structs that use image, without deviating from the existing function signatures and code structure.
 type Image interface {
 	IsClosed() bool
-	Poll(handler term.FragmentHandler, fragmentLimit int) (int, error)
-	BoundedPoll(handler term.FragmentHandler, limitPosition int64, fragmentLimit int) (int, error)
-	ControlledPoll(handler term.ControlledFragmentHandler, fragmentLimit int) (int, error)
+	Poll(handler term.FragmentHandler, fragmentLimit int) int
+	BoundedPoll(handler term.FragmentHandler, limitPosition int64, fragmentLimit int) int
+	ControlledPoll(handler term.ControlledFragmentHandler, fragmentLimit int) int
 	Position() int64
 	IsEndOfStream() bool
 	SessionID() int32

aeron/image_impl.go

@@ -17,7 +17,6 @@ limitations under the License.
 package aeron
 
 import (
-	"fmt"
 	"github.com/lirm/aeron-go/aeron/atomic"
 	"github.com/lirm/aeron-go/aeron/logbuffer"
 	"github.com/lirm/aeron-go/aeron/logbuffer/term"
@@ -77,23 +76,23 @@ func (image *image) IsClosed() bool {
 // early, even if fragmentLimit has not been hit.
 //
 //go:norace
-func (image *image) Poll(handler term.FragmentHandler, fragmentLimit int) (int, error) {
+func (image *image) Poll(handler term.FragmentHandler, fragmentLimit int) int {
 	if image.IsClosed() {
-		return 0, nil
+		return 0
 	}
 
 	position := image.subscriberPosition.get()
 	termOffset := int32(position) & image.termLengthMask
 	index := indexByPosition(position, image.positionBitsToShift)
 	termBuffer := image.termBuffers[index]
 
-	offset, result, err := term.Read(termBuffer, termOffset, handler, fragmentLimit, &image.header)
+	offset, result := term.Read(termBuffer, termOffset, handler, fragmentLimit, &image.header)
 
 	newPosition := position + int64(offset-termOffset)
 	if newPosition > position {
 		image.subscriberPosition.set(newPosition)
 	}
-	return result, err
+	return result
 }
 
 // BoundedPoll polls for new messages in a stream. If new messages are found
@@ -108,9 +107,9 @@ func (image *image) BoundedPoll(
 	handler term.FragmentHandler,
 	limitPosition int64,
 	fragmentLimit int,
-) (int, error) {
+) int {
 	if image.IsClosed() {
-		return 0, nil
+		return 0
 	}
 
 	fragmentsRead := 0
@@ -129,11 +128,9 @@ func (image *image) BoundedPoll(
 	header := &image.header
 	header.Wrap(termBuffer.Ptr(), termBuffer.Capacity())
 
-	var err error
-	for fragmentsRead < fragmentLimit && offset < limitOffset && err == nil {
+	for fragmentsRead < fragmentLimit && offset < limitOffset {
 		length := logbuffer.GetFrameLength(termBuffer, offset)
 		if length <= 0 {
-			err = fmt.Errorf("invalid frameLength %d", length)
 			break
 		}
 
@@ -147,14 +144,14 @@ func (image *image) BoundedPoll(
 		fragmentsRead++
 		header.SetOffset(frameOffset)
 
-		err = handler(termBuffer, frameOffset+logbuffer.DataFrameHeader.Length,
+		handler(termBuffer, frameOffset+logbuffer.DataFrameHeader.Length,
 			length-logbuffer.DataFrameHeader.Length, header)
 	}
 	resultingPosition := initialPosition + int64(offset-initialOffset)
 	if resultingPosition > initialPosition {
 		image.subscriberPosition.set(resultingPosition)
 	}
-	return fragmentsRead, err
+	return fragmentsRead
 }
 
 // ControlledPoll polls for new messages in a stream. If new messages are found
@@ -168,9 +165,9 @@ func (image *image) BoundedPoll(
 func (image *image) ControlledPoll(
 	handler term.ControlledFragmentHandler,
 	fragmentLimit int,
-) (int, error) {
+) int {
 	if image.IsClosed() {
-		return 0, nil
+		return 0
 	}
 
 	fragmentsRead := 0
@@ -185,11 +182,9 @@ func (image *image) ControlledPoll(
 	header := &image.header
 	header.Wrap(termBuffer.Ptr(), termBuffer.Capacity())
 
-	var err error
 	for fragmentsRead < fragmentLimit && offset < capacity {
 		length := logbuffer.GetFrameLength(termBuffer, offset)
 		if length <= 0 {
-			err = fmt.Errorf("invalid frameLength %d", length)
 			break
 		}
 
@@ -223,7 +218,7 @@ func (image *image) ControlledPoll(
 	if resultingPosition > initialPosition {
 		image.subscriberPosition.set(resultingPosition)
 	}
-	return fragmentsRead, err
+	return fragmentsRead
 }
 
 // Position returns the position this image has been consumed to by the subscriber.

aeron/logbuffer/term/fragmenthandler.go

@@ -21,4 +21,4 @@ import (
 )
 
 // FragmentHandler is the main callback interface for received data
-type FragmentHandler func(buffer *atomic.Buffer, offset int32, length int32, header *logbuffer.Header) error
+type FragmentHandler func(buffer *atomic.Buffer, offset int32, length int32, header *logbuffer.Header)

aeron/logbuffer/term/reader.go

@@ -1,6 +1,5 @@
 /*
 Copyright 2016 Stanislav Liberman
-Copyright 2022 Steven Stern
 
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.
@@ -18,26 +17,22 @@ limitations under the License.
 package term
 
 import (
-	"fmt"
 	"github.com/lirm/aeron-go/aeron/atomic"
 	"github.com/lirm/aeron-go/aeron/logbuffer"
 	"github.com/lirm/aeron-go/aeron/util"
 )
 
 // Read will attempt to read the next frame from the term and invoke the callback if successful.
-// Method will return a tuple of new term offset, number of fragments read, and an error reading
-// fragments.  An error will interrupt reading before fragmentLimit, but any fragments read prior
-// to the error will still be processed, and that number will be returned with the error.
+// Method will return a tuple of new term offset and number of fragments read
 //
 //go:norace
 func Read(termBuffer *atomic.Buffer, termOffset int32, handler FragmentHandler, fragmentLimit int,
-	header *logbuffer.Header) (int32, int, error) {
+	header *logbuffer.Header) (int32, int) {
 
 	capacity := termBuffer.Capacity()
 
 	var fragmentsRead int
-	var err error
-	for fragmentsRead < fragmentLimit && termOffset < capacity && err == nil {
+	for fragmentsRead < fragmentLimit && termOffset < capacity {
 		frameLength := logbuffer.GetFrameLength(termBuffer, termOffset)
 		if frameLength <= 0 {
 			break
@@ -48,27 +43,26 @@ func Read(termBuffer *atomic.Buffer, termOffset int32, handler FragmentHandler,
 
 		if !logbuffer.IsPaddingFrame(termBuffer, fragmentOffset) {
 			header.Wrap(termBuffer.Ptr(), termBuffer.Capacity())
-			fragmentsRead++
 			header.SetOffset(fragmentOffset)
-			err = handler(termBuffer, fragmentOffset+logbuffer.DataFrameHeader.Length,
+			handler(termBuffer, fragmentOffset+logbuffer.DataFrameHeader.Length,
 				frameLength-logbuffer.DataFrameHeader.Length, header)
+
+			fragmentsRead++
 		}
 	}
 
-	return termOffset, fragmentsRead, err
+	return termOffset, fragmentsRead
 }
 
 // BoundedRead will attempt to read frames from the term up to the specified offsetLimit.
 // Method will return a tuple of new term offset and number of fragments read
 func BoundedRead(termBuffer *atomic.Buffer, termOffset int32, offsetLimit int32, handler FragmentHandler,
-	fragmentLimit int, header *logbuffer.Header) (int32, int, error) {
+	fragmentLimit int, header *logbuffer.Header) (int32, int) {
 
 	var fragmentsRead int
-	var err error
-	for fragmentsRead < fragmentLimit && termOffset < offsetLimit && err == nil {
+	for fragmentsRead < fragmentLimit && termOffset < offsetLimit {
 		frameLength := logbuffer.GetFrameLength(termBuffer, termOffset)
 		if frameLength <= 0 {
-			err = fmt.Errorf("invalid frameLength %d", frameLength)
 			break
 		}
 
@@ -77,12 +71,13 @@ func BoundedRead(termBuffer *atomic.Buffer, termOffset int32, offsetLimit int32,
 
 		if !logbuffer.IsPaddingFrame(termBuffer, fragmentOffset) {
 			header.Wrap(termBuffer.Ptr(), termBuffer.Capacity())
-			fragmentsRead++
 			header.SetOffset(fragmentOffset)
-			err = handler(termBuffer, fragmentOffset+logbuffer.DataFrameHeader.Length,
+			handler(termBuffer, fragmentOffset+logbuffer.DataFrameHeader.Length,
 				frameLength-logbuffer.DataFrameHeader.Length, header)
+
+			fragmentsRead++
 		}
 	}
 
-	return termOffset, fragmentsRead, err
+	return termOffset, fragmentsRead
 }