func (cc *ClientConductor) AddSubscription(channel string, streamID int32) int64 {
	logger.Debugf("AddSubscription: channel=%s, streamId=%d", channel, streamID)
	cc.verifyDriverIsActive()
	cc.adminLock.Lock()
	defer cc.adminLock.Unlock()
	now := time.Now().UnixNano()
	regID := cc.driverProxy.AddSubscription(channel, streamID)
	subState := new(subscriptionStateDefn)
	subState.Init(channel, regID, streamID, now)
	cc.subs = append(cc.subs, subState)
	return regID
}
AddSubscription sends the add subscription command through the driver proxy
package aeron
import (
	"errors"
	"fmt"
	"log"
	"runtime"
	"sync"
	"time"
	"github.com/lirm/aeron-go/aeron/atomic"
	"github.com/lirm/aeron-go/aeron/broadcast"
	ctr "github.com/lirm/aeron-go/aeron/counters"
	"github.com/lirm/aeron-go/aeron/driver"
	"github.com/lirm/aeron-go/aeron/idlestrategy"
	"github.com/lirm/aeron-go/aeron/logbuffer"
	logging "github.com/op/go-logging"
)
var RegistrationStatus = struct {
	AwaitingMediaDriver   int
	RegisteredMediaDriver int
	ErroredMediaDriver    int
}{
	0,
	1,
	2,
}
const (
	keepaliveTimeoutNS = 500 * int64(time.Millisecond)
	resourceTimeoutNS  = 1000 * int64(time.Millisecond)
)
type publicationStateDefn struct {
	regID                    int64
	origRegID                int64
	timeOfRegistration       int64
	streamID                 int32
	sessionID                int32
	posLimitCounterID        int32
	channelStatusIndicatorID int32
	errorCode                int32
	status                   int
	channel                  string
	errorMessage             string
	buffers                  *logbuffer.LogBuffers
	publication              *Publication
}
func (pub *publicationStateDefn) Init(channel string, regID int64, streamID int32, now int64) *publicationStateDefn {
	pub.channel = channel
	pub.regID = regID
	pub.streamID = streamID
	pub.sessionID = -1
	pub.posLimitCounterID = -1
	pub.timeOfRegistration = now
	pub.status = RegistrationStatus.AwaitingMediaDriver
	return pub
}
type subscriptionStateDefn struct {
	regID              int64
	timeOfRegistration int64
	streamID           int32
	errorCode          int32
	status             int
	channel            string
	errorMessage       string
	subscription       *Subscription
}
func (sub *subscriptionStateDefn) Init(ch string, regID int64, sID int32, now int64) *subscriptionStateDefn {
	sub.channel = ch
	sub.regID = regID
	sub.streamID = sID
	sub.timeOfRegistration = now
	sub.status = RegistrationStatus.AwaitingMediaDriver
	return sub
}
type lingerResourse struct {
	lastTime int64
	resource *Image
}
type ClientConductor struct {
	pubs []*publicationStateDefn
	subs []*subscriptionStateDefn
	driverProxy *driver.Proxy
	counterValuesBuffer *atomic.Buffer
	counterReader       *ctr.Reader
	driverListenerAdapter *driver.ListenerAdapter
	adminLock sync.Mutex
	pendingCloses      map[int64]chan bool
	lingeringResources chan lingerResourse
	onNewPublicationHandler   NewPublicationHandler
	onNewSubscriptionHandler  NewSubscriptionHandler
	onAvailableImageHandler   AvailableImageHandler
	onUnavailableImageHandler UnavailableImageHandler
	errorHandler              func(error)
	running          atomic.Bool
	conductorRunning atomic.Bool
	driverActive     atomic.Bool
	timeOfLastKeepalive             int64
	timeOfLastCheckManagedResources int64
	timeOfLastDoWork                int64
	driverTimeoutNs                 int64
	interServiceTimeoutNs           int64
	publicationConnectionTimeoutNs  int64
	resourceLingerTimeoutNs         int64
}
func (cc *ClientConductor) Init(driverProxy *driver.Proxy, bcast *broadcast.CopyReceiver,
	interServiceTo, driverTo, pubConnectionTo, lingerTo time.Duration, counters *ctr.MetaDataFlyweight) *ClientConductor {
	logger.Debugf("Initializing ClientConductor with: %v %v %d %d %d", driverProxy, bcast, interServiceTo,
		driverTo, pubConnectionTo)
	cc.driverProxy = driverProxy
	cc.running.Set(true)
	cc.driverActive.Set(true)
	cc.driverListenerAdapter = driver.NewAdapter(cc, bcast)
	cc.interServiceTimeoutNs = interServiceTo.Nanoseconds()
	cc.driverTimeoutNs = driverTo.Nanoseconds()
	cc.publicationConnectionTimeoutNs = pubConnectionTo.Nanoseconds()
	cc.resourceLingerTimeoutNs = lingerTo.Nanoseconds()
	cc.counterValuesBuffer = counters.ValuesBuf.Get()
	cc.counterReader = ctr.NewReader(counters.ValuesBuf.Get(), counters.MetaDataBuf.Get())
	cc.pendingCloses = make(map[int64]chan bool)
	cc.lingeringResources = make(chan lingerResourse, 1024)
	cc.pubs = make([]*publicationStateDefn, 0)
	cc.subs = make([]*subscriptionStateDefn, 0)
	return cc
}
func (cc *ClientConductor) Close() error {
	logger.Debugf("Closing ClientConductor")
	var err error
	if cc.running.CompareAndSet(true, false) {
		for _, pub := range cc.pubs {
			if pub.publication != nil {
				err = pub.publication.Close()
				if err != nil {
					cc.errorHandler(err)
				}
			}
		}
		for _, sub := range cc.subs {
			if sub.subscription != nil {
				err = sub.subscription.Close()
				if err != nil {
					cc.errorHandler(err)
				}
			}
		}
	}
	timeoutDuration := 5 * time.Second
	timeout := time.Now().Add(timeoutDuration)
	for cc.conductorRunning.Get() && time.Now().Before(timeout) {
		time.Sleep(10 * time.Millisecond)
	}
	if cc.conductorRunning.Get() {
		msg := fmt.Sprintf("failed to stop conductor after %v", timeoutDuration)
		logger.Warning(msg)
		err = errors.New(msg)
	}
	logger.Debugf("Closed ClientConductor")
	return err
}
func (cc *ClientConductor) Start(idleStrategy idlestrategy.Idler) {
	cc.running.Set(true)
	go cc.run(idleStrategy)
}
func (cc *ClientConductor) run(idleStrategy idlestrategy.Idler) {
	now := time.Now().UnixNano()
	cc.timeOfLastKeepalive = now
	cc.timeOfLastCheckManagedResources = now
	cc.timeOfLastDoWork = now
	runtime.LockOSThread()
	defer func() {
		if err := recover(); err != nil {
			errStr := fmt.Sprintf("Panic: %v", err)
			logger.Error(errStr)
			cc.errorHandler(errors.New(errStr))
			cc.running.Set(false)
		}
		cc.conductorRunning.Set(false)
		logger.Infof("ClientConductor done")
	}()
	cc.conductorRunning.Set(true)
	for cc.running.Get() {
		workCount := cc.driverListenerAdapter.ReceiveMessages()
		workCount += cc.onHeartbeatCheckTimeouts()
		idleStrategy.Idle(workCount)
	}
}
func (cc *ClientConductor) verifyDriverIsActive() {
	if !cc.driverActive.Get() {
		log.Fatal("Driver is not active")
	}
}
func (cc *ClientConductor) AddPublication(channel string, streamID int32) int64 {
	logger.Debugf("AddPublication: channel=%s, streamId=%d", channel, streamID)
	cc.verifyDriverIsActive()
	cc.adminLock.Lock()
	defer cc.adminLock.Unlock()
	for _, pub := range cc.pubs {
		if pub.streamID == streamID && pub.channel == channel {
			return pub.regID
		}
	}
	now := time.Now().UnixNano()
	regID := cc.driverProxy.AddPublication(channel, streamID)
	pubState := new(publicationStateDefn)
	pubState.Init(channel, regID, streamID, now)
	cc.pubs = append(cc.pubs, pubState)
	return regID
}
func (cc *ClientConductor) AddExclusivePublication(channel string, streamID int32) int64 {
	logger.Debugf("AddExclusivePublication: channel=%s, streamId=%d", channel, streamID)
	cc.verifyDriverIsActive()
	cc.adminLock.Lock()
	defer cc.adminLock.Unlock()
	for _, pub := range cc.pubs {
		if pub.streamID == streamID && pub.channel == channel {
			return pub.regID
		}
	}
	now := time.Now().UnixNano()
	regID := cc.driverProxy.AddExclusivePublication(channel, streamID)
	pubState := new(publicationStateDefn)
	pubState.Init(channel, regID, streamID, now)
	cc.pubs = append(cc.pubs, pubState)
	return regID
}
func (cc *ClientConductor) FindPublication(regID int64) *Publication {
	cc.adminLock.Lock()
	defer cc.adminLock.Unlock()
	var publication *Publication
	for _, pub := range cc.pubs {
		if pub.regID == regID {
			if pub.publication != nil {
				publication = pub.publication
			} else {
				switch pub.status {
				case RegistrationStatus.AwaitingMediaDriver:
					waitForMediaDriver(pub.timeOfRegistration, cc)
				case RegistrationStatus.RegisteredMediaDriver:
					publication = NewPublication(pub.buffers)
					publication.conductor = cc
					publication.channel = pub.channel
					publication.regID = regID
					publication.originalRegID = pub.origRegID
					publication.streamID = pub.streamID
					publication.sessionID = pub.sessionID
					publication.pubLimit = NewPosition(cc.counterValuesBuffer, pub.posLimitCounterID)
					publication.channelStatusIndicatorID = pub.channelStatusIndicatorID
				case RegistrationStatus.ErroredMediaDriver:
					log.Fatalf("Error on %d: %d: %s", regID, pub.errorCode, pub.errorMessage)
				}
			}
			break
		}
	}
	return publication
}
func (cc *ClientConductor) releasePublication(regID int64) {
	logger.Debugf("ReleasePublication: regID=%d", regID)
	cc.verifyDriverIsActive()
	cc.adminLock.Lock()
	defer cc.adminLock.Unlock()
	pubcnt := len(cc.pubs)
	for i, pub := range cc.pubs {
		if pub != nil && pub.regID == regID {
			cc.driverProxy.RemovePublication(regID)
			cc.pubs[i] = cc.pubs[pubcnt-1]
			cc.pubs[pubcnt-1] = nil
			pubcnt--
		}
	}
	cc.pubs = cc.pubs[:pubcnt]
}