<commit_before>package proxy

import (
	"context"
	"fmt"
	"log"
	"sync"
	"time"

	sbAudio "github.com/dh1tw/remoteAudio/sb_audio"
	"github.com/gogo/protobuf/proto"
	"github.com/micro/go-micro/broker"
	"github.com/micro/go-micro/client"
)

type AudioServer struct {
	sync.RWMutex
	name           string
	serviceName    string
	client         client.Client
	rpc            sbAudio.ServerClient
	stateSub       broker.Subscriber
	rxAddress      string
	txAddress      string
	stateAddress   string
	rxOn           bool
	txUser         string
	latency        int
	notifyChangeCb func()
	closePing      chan struct{}
	doneCh         chan struct{}
	doneOnce       sync.Once
}

func NewAudioServer(name string, client client.Client, doneCh chan struct{}, opts ...Option) (*AudioServer, error) {

	serviceName := fmt.Sprintf("shackbus.radio.%s.audio", name)

	as := &AudioServer{
		name:         name,
		serviceName:  serviceName,
		client:       client,
		rxAddress:    fmt.Sprintf("%s.rx", serviceName),
		txAddress:    fmt.Sprintf("%s.tx", serviceName),
		stateAddress: fmt.Sprintf("%s.state", serviceName),
		txUser:       "",
		closePing:    make(chan struct{}),
		doneCh:       doneCh,
	}

	as.rpc = sbAudio.NewServerClient(as.serviceName, as.client)

	if err := as.getCapabilities(); err != nil {
		return nil, err
	}
	if err := as.getState(); err != nil {
		return nil, err
	}

	sub, err := as.client.Options().Broker.Subscribe(as.stateAddress, as.stateUpdateCb)
	if err != nil {
		return nil, err
	}

	as.stateSub = sub

	// start a go routine to ping our service every second
	go func() {
		for {
			select {
			case <-time.After(time.Second * 3):

				ping, err := as.ping()

				if err != nil {
					log.Println("unable to ping service", as.Name())
					as.closeDone()
					return
				}
				as.Lock()
				as.latency = ping
				as.Unlock()
				go as.notifyChangeCb()
			case <-as.closePing:
				return
			}
		}
	}()

	return as, nil
}

// the doneCh must be closed through this function to avoid
// multiple times closing this channel. Closing the doneCh signals the
// application that this object can be disposed.
func (as *AudioServer) closeDone() {
	as.doneOnce.Do(func() { close(as.doneCh) })
}

// Close shutsdown this object and all associated go routines so that
// it can be garbage collected.
func (as *AudioServer) Close() {
	as.Lock()
	defer as.Unlock()

	as.stateSub.Unsubscribe()
	close(as.closePing)
	as.rpc = nil
	as.client = nil
	as.notifyChangeCb = nil
	as.closeDone()
}

func (as *AudioServer) ping() (int, error) {

	ping := time.Now().UnixNano() / int64(time.Millisecond)

	pingMsg := sbAudio.PingPong{
		Ping: ping,
	}
<commit_msg>updated sbAudio ServerService<commit_after>package proxy

import (
	"context"
	"fmt"
	"log"
	"sync"
	"time"

	sbAudio "github.com/dh1tw/remoteAudio/sb_audio"
	"github.com/gogo/protobuf/proto"
	"github.com/micro/go-micro/broker"
	"github.com/micro/go-micro/client"
)

type AudioServer struct {
	sync.RWMutex
	name           string
	serviceName    string
	client         client.Client
	rpc            sbAudio.ServerService
	stateSub       broker.Subscriber
	rxAddress      string
	txAddress      string
	stateAddress   string
	rxOn           bool
	txUser         string
	latency        int
	notifyChangeCb func()
	closePing      chan struct{}
	doneCh         chan struct{}
	doneOnce       sync.Once
}

func NewAudioServer(name string, client client.Client, doneCh chan struct{}, opts ...Option) (*AudioServer, error) {

	serviceName := fmt.Sprintf("shackbus.radio.%s.audio", name)

	as := &AudioServer{
		name:         name,
		serviceName:  serviceName,
		client:       client,
		rxAddress:    fmt.Sprintf("%s.rx", serviceName),
		txAddress:    fmt.Sprintf("%s.tx", serviceName),
		stateAddress: fmt.Sprintf("%s.state", serviceName),
		txUser:       "",
		closePing:    make(chan struct{}),
		doneCh:       doneCh,
	}

	as.rpc = sbAudio.NewServerClient(as.serviceName, as.client)

	if err := as.getCapabilities(); err != nil {
		return nil, err
	}
	if err := as.getState(); err != nil {
		return nil, err
	}

	sub, err := as.client.Options().Broker.Subscribe(as.stateAddress, as.stateUpdateCb)
	if err != nil {
		return nil, err
	}

	as.stateSub = sub

	// start a go routine to ping our service every second
	go func() {
		for {
			select {
			case <-time.After(time.Second * 3):

				ping, err := as.ping()

				if err != nil {
					log.Println("unable to ping service", as.Name())
					as.closeDone()
					return
				}
				as.Lock()
				as.latency = ping
				as.Unlock()
				go as.notifyChangeCb()
			case <-as.closePing:
				return
			}
		}
	}()

	return as, nil
}

// the doneCh must be closed through this function to avoid
// multiple times closing this channel. Closing the doneCh signals the
// application that this object can be disposed.
func (as *AudioServer) closeDone() {
	as.doneOnce.Do(func() { close(as.doneCh) })
}

// Close shutsdown this object and all associated go routines so that
// it can be garbage collected.
func (as *AudioServer) Close() {
	as.Lock()
	defer as.Unlock()

	as.stateSub.Unsubscribe()
	close(as.closePing)
	as.rpc = nil
	as.client = nil
	as.notifyChangeCb = nil
	as.closeDone()
}

// ping performs a ping request to the audio server and returns the
// oneway latency (ping/2) in milli seconds.
func (as *AudioServer) ping() (int, error) {

	ping := time.Now().UnixNano() / int64(time.Millisecond)

	pingMsg := sbAudio.PingPong{
		Ping: ping,
	}

{'fraction_non_alphanumeric': 0.12385152517456817, 'fraction_numerical': 0.0016538037486218302, 'mean_word_length': 2.913012221423436, 'pattern_counts': {'":': 0, '<': 7, '<?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': '454017', 'n_tokens_mistral': 2021, 'n_tokens_neox': 1841, 'n_words': 496}