Merge pull request #46 from billsegall/archive-beta2

Archive beta2

Author: lirm

aeron/image.go

@@ -118,6 +118,25 @@ func (image *Image) Poll(handler term.FragmentHandler, fragmentLimit int) int {
 	return result
 }
 
+func (image *Image) PollWithContext(handler term.FragmentHandler, fragmentLimit int) int {
+	if image.IsClosed() {
+		return 0
+	}
+
+	position := image.subscriberPosition.get()
+	termOffset := int32(position) & image.termLengthMask
+	index := indexByPosition(position, image.positionBitsToShift)
+	termBuffer := image.termBuffers[index]
+
+	offset, result := term.ReadWithContext(termBuffer, termOffset, handler, fragmentLimit, &image.header)
+
+	newPosition := position + int64(offset-termOffset)
+	if newPosition > position {
+		image.subscriberPosition.set(newPosition)
+	}
+	return result
+}
+
 // Position returns the position this Image has been consumed to by the subscriber.
 func (image *Image) Position() int64 {
 	if image.IsClosed() {

aeron/logbuffer/term/reader.go

@@ -54,3 +54,32 @@ func Read(termBuffer *atomic.Buffer, termOffset int32, handler FragmentHandler,
 
 	return termOffset, fragmentsRead
 }
+
+// ReadWithContext as for Read() but woth a contextual argument
+func ReadWithContext(termBuffer *atomic.Buffer, termOffset int32, handler FragmentHandler, fragmentsLimit int,
+	header *logbuffer.Header) (int32, int) {
+
+	capacity := termBuffer.Capacity()
+
+	var fragmentsRead int
+	for fragmentsRead < fragmentsLimit && termOffset < capacity {
+		frameLength := logbuffer.GetFrameLength(termBuffer, termOffset)
+		if frameLength <= 0 {
+			break
+		}
+
+		fragmentOffset := termOffset
+		termOffset += util.AlignInt32(frameLength, logbuffer.FrameAlignment)
+
+		if !logbuffer.IsPaddingFrame(termBuffer, fragmentOffset) {
+			header.Wrap(termBuffer.Ptr(), termBuffer.Capacity())
+			header.SetOffset(fragmentOffset)
+			handler(termBuffer, fragmentOffset+logbuffer.DataFrameHeader.Length,
+				frameLength-logbuffer.DataFrameHeader.Length, header)
+
+			fragmentsRead++
+		}
+	}
+
+	return termOffset, fragmentsRead
+}

aeron/subscription.go

@@ -102,6 +102,33 @@ func (sub *Subscription) Poll(handler term.FragmentHandler, fragmentLimit int) i
 	return fragmentsRead
 }
 
+// PollWithContext as for Poll() but provides an integer argument for passing contextual information
+func (sub *Subscription) PollWithContext(handler term.FragmentHandler, fragmentLimit int) int {
+
+	img := sub.images.Get()
+	length := len(img)
+	var fragmentsRead int
+
+	if length > 0 {
+		startingIndex := sub.roundRobinIndex
+		sub.roundRobinIndex++
+		if startingIndex >= length {
+			sub.roundRobinIndex = 0
+			startingIndex = 0
+		}
+
+		for i := startingIndex; i < length && fragmentsRead < fragmentLimit; i++ {
+			fragmentsRead += img[i].PollWithContext(handler, fragmentLimit-fragmentsRead)
+		}
+
+		for i := 0; i < startingIndex && fragmentsRead < fragmentLimit; i++ {
+			fragmentsRead += img[i].PollWithContext(handler, fragmentLimit-fragmentsRead)
+		}
+	}
+
+	return fragmentsRead
+}
+
 func (sub *Subscription) hasImage(sessionID int32) bool {
 	img := sub.images.Get()
 	for _, image := range img {
@@ -144,6 +171,16 @@ func (sub *Subscription) RegistrationID() int64 {
 	return sub.registrationID
 }
 
+// IsConnected returns if this subscription is connected by having at least one open publication Image.
+func (sub *Subscription) IsConnected() bool {
+	for _, image := range sub.images.Get() {
+		if !image.IsClosed() {
+			return true
+		}
+	}
+	return false
+}
+
 // HasImages is a helper method checking whether this subscription has any images associated with it.
 func (sub *Subscription) HasImages() bool {
 	images := sub.images.Get()

archive/README.md

@@ -7,7 +7,7 @@ protocol](http://github.com/real-logic/aeron/blob/master/aeron-archive/src/main/
 is specified in xml using the [Simple Binary Encoding (SBE)](https://github.com/real-logic/simple-binary-encoding)
 
 ## Current State
-The implementation is the first beta release. The API will be changed only if required for bugfixes.
+The implementation is the second beta release. The API may still be changed for bug fixes or significant issues.
 
 # Design
 
@@ -22,21 +22,21 @@ the archive library is a layering on top of that.
 
 Finally golang idioms are used where reasonable.
 
-The archive library does not lock and concurrent calls to archive
-library calls that invoke the aeron-archive protocol calls should be
-externally locked to ensure only one concuurrent access.
+The archive library does not lock and concurrent calls on an archive
+client to invoke the aeron-archive protocol calls should be externally
+locked to ensure only one concurrent access.
 
 ### Naming and other choices
 
 Function names used in archive.go which contains the main API are
 based on the Java names so that developers can more easily switch
-between langugages and so that any API documentation is more useful
+between languages and so that any API documentation is more useful
 across implementations. Some differences exist due to capitalization
 requirements, lack of polymorphism, etc.
 
 Function names used in encoders.go and proxy.go are based on the
-protocol specification. Where the protocol specifies a type that cab
-ne naturally repreented in golang, the golang type is used used where
+protocol specification. Where the protocol specifies a type that can
+be naturally represented in golang, the golang type is used used where
 possible until encoding. Examples include the use of `bool` rather than
 `BooleanType` and `string` over `[]uint8`
 
@@ -72,28 +72,40 @@ operations in progress when polling.
 
 ## Examples
 
-Examples are provided for a [basic_recording_publisher](examples/basic_recording_publisher/basic_recording_publisher.go) and [basic_replayed_subscriber](examples/basic_replayed_subscriber/basic_replayed_subscriber.go) that interoperate with the Java examples
+Examples are provided for a [basic_recording_publisher](examples/basic_recording_publisher/basic_recording_publisher.go) and [basic_replayed_subscriber](examples/basic_replayed_subscriber/basic_replayed_subscriber.go) that interoperate with the Java examples.
 
 ## Security
 
 Enabling security is done via setting the various auth options. [config_test.go](config_test.go) and [archive_test.go](archive_test.go) provide an example.
 
-The actual semantics of the security are dependent upon which authenticator aupplier you use and is tested agains [secure-logging-archiving-media-driver](secure-logging-archiving-media-driver).
+The actual semantics of the security are dependent upon which authenticator supplier you use and is tested against [secure-logging-archiving-media-driver](secure-logging-archiving-media-driver).
 
 # Backlog
  * godoc improvements
  * more testing
   * archive-media-driver mocking/execution
   * test cleanup in the media driver can be problematic
- * The archive state is largely unused. 
-   * Add? and use? IsOpen()
  * Auth should provide some callout mechanism
  * various FIXMEs
 
 # Bigger picture issues
  * Decided not to do locking in sync api, could subsequently add locks, or just async with locks.
    It may be that the control marshaller should be parameterized for this.
  * Java and C++ poll the counters to determine when a recording has actually started but the counters are not
-   availabe in go. As a result we use delays and 'hope' which isn't ideal.
+   available in go. As a result we use delays and 'hope' which isn't ideal.
  * It would be nice to silence the OnAvailableCounter noise
  * Within aeron-go there are cases of Log.Fatalf(), see for example trying to add a publication on a "bogus" channel.
+
+## Release Notes
+
+### 1.0b2
+ * Handle different archive clients using same channel/stream pairing
+ * Provide Subscription.IsConnected() and IsRecordingEventsConnected()
+ * Replace go-logging with zap to avoid reentrancy crashes in logging library
+ * Improve the logging by downgrading in severity and message tone some warning level messages
+ * Fix a race condition looking up correlationIDs on ControlResponse
+ * Fix a return code error in StopRecordingById()
+ * Fix unused argumentin StopRecording()
+ * Cosmetic improvements for golint and staticcheck
+ * Make the Listeners used for async events be per archive client instead of global
+