<commit_before>package trx

import (
	"fmt"
	"log"
	"sync"

	"github.com/dh1tw/remoteAudio/audio/pbWriter"

	"github.com/dh1tw/remoteAudio/audio/pbReader"

	"github.com/dh1tw/remoteAudio/audio/chain"
	"github.com/dh1tw/remoteAudio/proxy"
	"github.com/micro/go-micro/broker"
)

type Trx struct {
	sync.RWMutex
	rx                   *chain.Chain
	tx                   *chain.Chain
	fromNetwork          *pbReader.PbReader
	toNetwork            *pbWriter.PbWriter
	broker               broker.Broker
	servers              map[string]*proxy.AudioServer
	rxAudioSub           broker.Subscriber
	curServer            *proxy.AudioServer
	notifyServerChangeCb func()
}

type Options struct {
	Rx          *chain.Chain
	Tx          *chain.Chain
	FromNetwork *pbReader.PbReader
	ToNetwork   *pbWriter.PbWriter
	Broker      broker.Broker
}

func NewTrx(opts Options) (*Trx, error) {

	if opts.Rx == nil {
		return nil, fmt.Errorf("rx variable is nil")
	}
	if opts.Tx == nil {
		return nil, fmt.Errorf("tx variable is nil")
	}
	if opts.FromNetwork == nil {
		return nil, fmt.Errorf("fromNetwork sink is nil")
	}
	if opts.ToNetwork == nil {
		return nil, fmt.Errorf("toNetwork sink is nil")
	}
	if opts.Broker == nil {
		return nil, fmt.Errorf("broker is nil")
	}

	trx := &Trx{
		rx:          opts.Rx,
		tx:          opts.Tx,
		fromNetwork: opts.FromNetwork,
		toNetwork:   opts.ToNetwork,
		broker:      opts.Broker,
		servers:     make(map[string]*proxy.AudioServer),
	}

	trx.toNetwork.SetToWireCb(trx.toWireCb)

	return trx, nil
}

func (x *Trx) SetNotifyServerChangeCb(f func()) {
	x.Lock()
	defer x.Unlock()
	x.notifyServerChangeCb = f
}

func (x *Trx) AddServer(asvr *proxy.AudioServer) {
	x.Lock()
	defer x.Unlock()

	if asvr == nil {
		return
	}
	_, ok := x.servers[asvr.Name()]
	if ok {
		return
	}
	x.servers[asvr.Name()] = asvr

	asvr.SetNotifyCb(x.onAudioServersChanged)
	go x.onAudioServersChanged()

	log.Println("added audio server", asvr.Name())
}

func (x *Trx) onAudioServersChanged() {
	x.RLock()
	defer x.RUnlock()
	if x.notifyServerChangeCb != nil {
		x.notifyServerChangeCb()
	}
}

// Server returns a particular AudioServer. If no
// AudioServer exists with that name, (nil, false) will be returned.
func (x *Trx) Server(name string) (*proxy.AudioServer, bool) {
	x.RLock()
	defer x.RUnlock()

	svr, ok := x.servers[name]
	return svr, ok
}

// Servers returns a list with the names of the registered
// audio servers.
func (x *Trx) Servers() []string {
	x.RLock()
	defer x.RUnlock()

	list := []string{}
	for _, aServer := range x.servers {
		list = append(list, aServer.Name())
	}
	return list
}

func (x *Trx) RemoveServer(asName string) {<commit_msg>call notifyServerChangedCb async to avoid deadlock<commit_after>package trx

import (
	"fmt"
	"log"
	"sync"

	"github.com/dh1tw/remoteAudio/audio/pbWriter"

	"github.com/dh1tw/remoteAudio/audio/pbReader"

	"github.com/dh1tw/remoteAudio/audio/chain"
	"github.com/dh1tw/remoteAudio/proxy"
	"github.com/micro/go-micro/broker"
)

type Trx struct {
	sync.RWMutex
	rx                   *chain.Chain
	tx                   *chain.Chain
	fromNetwork          *pbReader.PbReader
	toNetwork            *pbWriter.PbWriter
	broker               broker.Broker
	servers              map[string]*proxy.AudioServer
	rxAudioSub           broker.Subscriber
	curServer            *proxy.AudioServer
	notifyServerChangeCb func()
}

type Options struct {
	Rx          *chain.Chain
	Tx          *chain.Chain
	FromNetwork *pbReader.PbReader
	ToNetwork   *pbWriter.PbWriter
	Broker      broker.Broker
}

func NewTrx(opts Options) (*Trx, error) {

	if opts.Rx == nil {
		return nil, fmt.Errorf("rx variable is nil")
	}
	if opts.Tx == nil {
		return nil, fmt.Errorf("tx variable is nil")
	}
	if opts.FromNetwork == nil {
		return nil, fmt.Errorf("fromNetwork sink is nil")
	}
	if opts.ToNetwork == nil {
		return nil, fmt.Errorf("toNetwork sink is nil")
	}
	if opts.Broker == nil {
		return nil, fmt.Errorf("broker is nil")
	}

	trx := &Trx{
		rx:          opts.Rx,
		tx:          opts.Tx,
		fromNetwork: opts.FromNetwork,
		toNetwork:   opts.ToNetwork,
		broker:      opts.Broker,
		servers:     make(map[string]*proxy.AudioServer),
	}

	trx.toNetwork.SetToWireCb(trx.toWireCb)

	return trx, nil
}

func (x *Trx) SetNotifyServerChangeCb(f func()) {
	x.Lock()
	defer x.Unlock()
	x.notifyServerChangeCb = f
}

func (x *Trx) AddServer(asvr *proxy.AudioServer) {
	x.Lock()
	defer x.Unlock()

	if asvr == nil {
		return
	}
	_, ok := x.servers[asvr.Name()]
	if ok {
		return
	}
	x.servers[asvr.Name()] = asvr

	asvr.SetNotifyCb(x.onAudioServersChanged)
	go x.onAudioServersChanged()

	log.Println("added audio server", asvr.Name())
}

func (x *Trx) onAudioServersChanged() {
	x.RLock()
	defer x.RUnlock()
	if x.notifyServerChangeCb != nil {
		go x.notifyServerChangeCb()
	}
}

// Server returns a particular AudioServer. If no
// AudioServer exists with that name, (nil, false) will be returned.
func (x *Trx) Server(name string) (*proxy.AudioServer, bool) {
	x.RLock()
	defer x.RUnlock()

	svr, ok := x.servers[name]
	return svr, ok
}

// Servers returns a list with the names of the registered
// audio servers.
func (x *Trx) Servers() []string {
	x.RLock()
	defer x.RUnlock()

	list := []string{}
	for _, aServer := range x.servers {
		list = append(list, aServer.Name())
	}
	return list
}

func (x *Trx) RemoveServer(asName string) {
{'fraction_non_alphanumeric': 0.12849162011173185, 'fraction_numerical': 0.0014897579143389199, 'mean_word_length': 3.0658591975775926, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '282930', 'n_tokens_mistral': 2030, 'n_tokens_neox': 1981, 'n_words': 464}