func (archive *Archive) StopRecordingByPublication(publication aeron.Publication) error {
	channel, err := AddSessionIdToChannel(publication.Channel(), publication.SessionID())
	if err != nil {
		return err
	}
	return archive.StopRecording(channel, publication.StreamID())
}
StopRecording by Publication

Stop recording a sessionId specific recording that pertains to the given Publication

Returns error on failure, nil on success
package archive
import (
	"fmt"
	"github.com/lirm/aeron-go/aeron"
	"github.com/lirm/aeron-go/aeron/atomic"
	"github.com/lirm/aeron-go/archive/codecs"
	logging "github.com/op/go-logging"
	"time"
)
type Archive struct {
	aeron        *aeron.Aeron             
	aeronContext *aeron.Context           
	Options      *Options                 
	SessionId    int64                    
	Proxy        *Proxy                   
	Control      *Control                 
	Events       *RecordingEventsAdapter  
}
const (
	RecordingPositionNull = int64(-1)         
	RecordingLengthNull   = int64(-1)         
	RecordingLengthMax    = int64(2<<31 - 1)  
)
const (
	RecordingIdNullValue = int32(-1)
)
var Listeners *ArchiveListeners
type ArchiveListeners struct {
	ErrorListener func(error)
	RecordingEventStartedListener  func(*codecs.RecordingStarted)
	RecordingEventProgressListener func(*codecs.RecordingProgress)
	RecordingEventStoppedListener  func(*codecs.RecordingStopped)
	RecordingSignalListener func(*codecs.RecordingSignalEvent)
	NewSubscriptionListener  func(string, int32, int64)
	NewPublicationListener   func(string, int32, int32, int64)
	AvailableImageListener   func(*aeron.Image)
	UnavailableImageListener func(*aeron.Image)
}
func LoggingErrorListener(err error) {
	logger.Errorf("Error: %s\n", err.Error())
}
func LoggingRecordingSignalListener(rse *codecs.RecordingSignalEvent) {
	logger.Infof("RecordingSignalListener, signal event is %#v\n", rse)
}
func LoggingRecordingEventStartedListener(rs *codecs.RecordingStarted) {
	logger.Infof("RecordingEventStartedListener: %#v\n", rs)
}
func LoggingRecordingEventProgressListener(rp *codecs.RecordingProgress) {
	logger.Infof("RecordingEventProgressListener, event is %#v\n", rp)
}
func LoggingRecordingEventStoppedListener(rs *codecs.RecordingStopped) {
	logger.Infof("RecordingEventStoppedListener, event is %#v\n", rs)
}
func LoggingNewSubscriptionListener(channel string, stream int32, correlationId int64) {
	logger.Infof("NewSubscriptionListener(channel:%s stream:%d correlationId:%d)\n", channel, stream, correlationId)
}
func LoggingNewPublicationListener(channel string, stream int32, session int32, regId int64) {
	logger.Infof("NewPublicationListener(channel:%s stream:%d, session:%d, regId:%d)", channel, stream, session, regId)
}
func LoggingAvailableImageListener(image *aeron.Image) {
	logger.Infof("NewAvailableImageListener, sessionId is %d\n", image.SessionID())
}
func LoggingUnavailableImageListener(image *aeron.Image) {
	logger.Infof("NewUnavalableImageListener, sessionId is %d\n", image.SessionID())
}
var rangeChecking bool
var logger = logging.MustGetLogger("archive")
var _correlationId atomic.Long
var sessionsMap map[int64]*Control      
var correlationsMap map[int64]*Control  
var recordingsMap map[int64]*Control    
func init() {
	_correlationId.Set(time.Now().UnixNano())
	sessionsMap = make(map[int64]*Control)
	correlationsMap = make(map[int64]*Control)
	recordingsMap = make(map[int64]*Control)
}
func nextCorrelationId() int64 {
	return _correlationId.Inc()
}
func ReplaySessionIdToStreamId(replaySessionId int64) int32 {
	return int32(replaySessionId)
}
func AddSessionIdToChannel(channel string, sessionId int32) (string, error) {
	uri, err := aeron.ParseChannelUri(channel)
	if err != nil {
		return channel, err
	}
	uri.Set("session-id", fmt.Sprint(sessionId))
	return uri.String(), nil
}
func NewArchive(options *Options, context *aeron.Context) (*Archive, error) {
	var err error
	archive := new(Archive)
	archive.aeron = new(aeron.Aeron)
	archive.aeronContext = context
	if options != nil {
		archive.Options = options
	} else {
		if archive.Options == nil {
			archive.Options = DefaultOptions()
		}
	}
	logging.SetLevel(archive.Options.ArchiveLoglevel, "archive")
	logging.SetLevel(archive.Options.AeronLoglevel, "aeron")
	logging.SetLevel(archive.Options.AeronLoglevel, "memmap")
	logging.SetLevel(archive.Options.AeronLoglevel, "driver")
	logging.SetLevel(archive.Options.AeronLoglevel, "counters")
	logging.SetLevel(archive.Options.AeronLoglevel, "logbuffers")
	logging.SetLevel(archive.Options.AeronLoglevel, "buffer")
	logging.SetLevel(archive.Options.AeronLoglevel, "rb")
	archive.Control = new(Control)
	archive.Control.archive = archive
	archive.Proxy = new(Proxy)
	archive.Proxy.archive = archive
	archive.Proxy.marshaller = codecs.NewSbeGoMarshaller()
	archive.Events = new(RecordingEventsAdapter)
	archive.Events.archive = archive
	Listeners = new(ArchiveListeners)
	Listeners.ErrorListener = LoggingErrorListener
	archive.SetAeronErrorHandler(LoggingErrorListener)
	if logging.GetLevel("archive") >= logging.DEBUG {
		logger.Debugf("Setting logging listeners")
		Listeners.RecordingEventStartedListener = LoggingRecordingEventStartedListener
		Listeners.RecordingEventProgressListener = LoggingRecordingEventProgressListener
		Listeners.RecordingEventStoppedListener = LoggingRecordingEventStoppedListener
		Listeners.RecordingSignalListener = LoggingRecordingSignalListener
		Listeners.AvailableImageListener = LoggingAvailableImageListener
		Listeners.UnavailableImageListener = LoggingUnavailableImageListener
		Listeners.NewSubscriptionListener = LoggingNewSubscriptionListener
		Listeners.NewPublicationListener = LoggingNewPublicationListener
		archive.aeronContext.NewSubscriptionHandler(Listeners.NewSubscriptionListener)
		archive.aeronContext.NewPublicationHandler(Listeners.NewPublicationListener)
	}
	archive.aeron, err = aeron.Connect(archive.aeronContext)
	if err != nil {
		return archive, err
	}
	archive.Control.Subscription = <-archive.aeron.AddSubscription(archive.Options.ResponseChannel, archive.Options.ResponseStream)
	logger.Debugf("Control response subscription: %#v", archive.Control.Subscription)
	archive.Proxy.Publication = <-archive.aeron.AddExclusivePublication(archive.Options.RequestChannel, archive.Options.RequestStream)
	logger.Debugf("Proxy request publication: %#v", archive.Proxy.Publication)
	archive.Control.State.state = ControlStateConnectRequestSent
	correlationId := nextCorrelationId()
	correlationsMap[correlationId] = archive.Control  
	defer correlationsMapClean(correlationId)         
	if archive.Options.AuthEnabled {
		if err := archive.Proxy.AuthConnectRequest(correlationId, archive.Options.ResponseStream, archive.Options.ResponseChannel, archive.Options.AuthCredentials); err != nil {
			logger.Errorf("ConnectRequest failed: %s\n", err)
			return nil, err
		}
	} else {
		if err := archive.Proxy.ConnectRequest(correlationId, archive.Options.ResponseStream, archive.Options.ResponseChannel); err != nil {
			logger.Errorf("ConnectRequest failed: %s\n", err)
			return nil, err
		}
	}
	start := time.Now()
	for archive.Control.State.state != ControlStateConnected && archive.Control.State.err == nil {
		fragments := archive.Control.Poll(ConnectionControlFragmentHandler, 1)
		if fragments > 0 {
			logger.Debugf("Read %d fragment(s)\n", fragments)
		}
		if time.Since(start) > archive.Options.Timeout {
			archive.Control.State.state = ControlStateTimedOut
			archive.Control.State.err = fmt.Errorf("Operation timed out")
			break
		} else {
			archive.Options.IdleStrategy.Idle(0)
		}
	}
	if archive.Control.State.err != nil {
		logger.Errorf("Connect failed: %s\n", archive.Control.State.err)
	} else if archive.Control.State.state != ControlStateConnected {
		logger.Error("Connect failed\n")
	} else {
		logger.Infof("Archive connection established for sessionId:%d\n", archive.SessionId)
		sessionsMap[archive.SessionId] = archive.Control  
	}
	return archive, archive.Control.State.err
}
func (archive *Archive) Close() error {
	archive.Proxy.CloseSessionRequest()
	archive.Proxy.Publication.Close()
	archive.Control.Subscription.Close()
	delete(sessionsMap, archive.SessionId)
	return archive.aeron.Close()
}
func (archive *Archive) SetAeronErrorHandler(handler func(error)) {
	archive.aeronContext.ErrorHandler(handler)
}
func (archive *Archive) SetAeronDir(dir string) {
	archive.aeronContext.AeronDir(dir)
}
func (archive *Archive) SetAeronMediaDriverTimeout(timeout time.Duration) {
	archive.aeronContext.MediaDriverTimeout(timeout)
}
func (archive *Archive) SetAeronResourceLingerTimeout(timeout time.Duration) {
	archive.aeronContext.ResourceLingerTimeout(timeout)
}
func (archive *Archive) SetAeronInterServiceTimeout(timeout time.Duration) {
	archive.aeronContext.InterServiceTimeout(timeout)
}
func (archive *Archive) SetAeronPublicationConnectionTimeout(timeout time.Duration) {
	archive.aeronContext.PublicationConnectionTimeout(timeout)
}
func (archive *Archive) AeronCncFileName() string {
	return archive.aeronContext.CncFileName()
}
func (archive *Archive) EnableRecordingEvents() {
	archive.Events.Subscription = <-archive.aeron.AddSubscription(archive.Options.RecordingEventsChannel, archive.Options.RecordingEventsStream)
	archive.Events.Enabled = true
	logger.Debugf("RecordingEvents subscription: %#v", archive.Events.Subscription)
}
func (archive *Archive) DisableRecordingEvents() {
	archive.Events.Subscription.Close()
	archive.Events.Enabled = false
	logger.Debugf("RecordingEvents subscription closed")
}
func (archive *Archive) RecordingEventsPoll() int {
	return archive.Events.Poll(nil, 1)
}
func (archive *Archive) AddSubscription(channel string, streamId int32) chan *aeron.Subscription {
	return archive.aeron.AddSubscription(channel, streamId)
}
func (archive *Archive) AddPublication(channel string, streamId int32) chan *aeron.Publication {
	return archive.aeron.AddPublication(channel, streamId)
}
func (archive *Archive) AddExclusivePublication(channel string, streamId int32) chan *aeron.Publication {
	return archive.aeron.AddExclusivePublication(channel, streamId)
}
func (archive *Archive) ClientId() int64 {
	return archive.aeron.ClientID()
}
func correlationsMapClean(correlationId int64) {
	delete(correlationsMap, correlationId)
}
func (archive *Archive) StartRecording(channel string, stream int32, isLocal bool, autoStop bool) (int64, error) {
	logger.Debugf("StartRecording(%s:%d)\n", channel, stream)
	correlationId := nextCorrelationId()
	correlationsMap[correlationId] = archive.Control  
	defer correlationsMapClean(correlationId)         
	if err := archive.Proxy.StartRecordingRequest(correlationId, stream, isLocal, autoStop, channel); err != nil {
		return 0, err
	}
	return archive.Control.PollForResponse(correlationId)
}
func (archive *Archive) StopRecording(channel string, stream int32) error {
	logger.Debugf("StopRecording(%s:%d)\n", channel, stream)
	correlationId := nextCorrelationId()
	correlationsMap[correlationId] = archive.Control  
	defer correlationsMapClean(correlationId)         
	if err := archive.Proxy.StopRecordingRequest(correlationId, stream, channel); err != nil {
		return err
	}
	_, err := archive.Control.PollForResponse(correlationId)
	return err
}
func (archive *Archive) StopRecordingByIdentity(recordingId int64) (bool, error) {
	logger.Debugf("StopRecordingByIdentity(%d)\n", recordingId)
	correlationId := nextCorrelationId()
	correlationsMap[correlationId] = archive.Control  
	defer correlationsMapClean(correlationId)         
	if err := archive.Proxy.StopRecordingByIdentityRequest(correlationId, recordingId); err != nil {
		return false, err
	}
	res, err := archive.Control.PollForResponse(correlationId)
	if res != 0 {
		logger.Debugf("StopRecordingByIdentity result was %d\n", res)
	}
	return res == 0, err
}
func (archive *Archive) StopRecordingBySubscriptionId(subscriptionId int64) error {
	logger.Debugf("StopRecordingBySubscriptionId(%d)\n", subscriptionId)
	correlationId := nextCorrelationId()
	correlationsMap[correlationId] = archive.Control  
	defer correlationsMapClean(correlationId)         
	if err := archive.Proxy.StopRecordingSubscriptionRequest(correlationId, subscriptionId); err != nil {
		return err
	}
	_, err := archive.Control.PollForResponse(correlationId)
	return err
}