Use new go.19 atomic[T]
This commit is contained in:
parent
d919c27fd0
commit
351897ba56
28
listener.go
28
listener.go
|
@ -30,13 +30,11 @@ func (l *listener) Accept() (net.Conn, error) {
|
||||||
|
|
||||||
log.Printf("accepted new connection from %s\n", c.RemoteAddr().String())
|
log.Printf("accepted new connection from %s\n", c.RemoteAddr().String())
|
||||||
tc := &Conn{
|
tc := &Conn{
|
||||||
Conn: c,
|
Conn: c,
|
||||||
ReadTimeout: l.ReadTimeout,
|
ReadTimeout: l.ReadTimeout,
|
||||||
WriteTimeout: l.WriteTimeout,
|
WriteTimeout: l.WriteTimeout,
|
||||||
ReadThreshold: int32((l.ReadTimeout * 1024) / time.Second),
|
ReadThreshold: int32((l.ReadTimeout * 1024) / time.Second),
|
||||||
WriteThreshold: int32((l.WriteTimeout * 1024) / time.Second),
|
WriteThreshold: int32((l.WriteTimeout * 1024) / time.Second),
|
||||||
BytesReadFromDeadline: 0,
|
|
||||||
BytesWrittenFromDeadline: 0,
|
|
||||||
}
|
}
|
||||||
return tc, nil
|
return tc, nil
|
||||||
}
|
}
|
||||||
|
@ -49,13 +47,13 @@ type Conn struct {
|
||||||
WriteTimeout time.Duration
|
WriteTimeout time.Duration
|
||||||
ReadThreshold int32
|
ReadThreshold int32
|
||||||
WriteThreshold int32
|
WriteThreshold int32
|
||||||
BytesReadFromDeadline int32
|
BytesReadFromDeadline atomic.Int32
|
||||||
BytesWrittenFromDeadline int32
|
BytesWrittenFromDeadline atomic.Int32
|
||||||
}
|
}
|
||||||
|
|
||||||
func (c *Conn) Read(b []byte) (n int, err error) {
|
func (c *Conn) Read(b []byte) (n int, err error) {
|
||||||
if atomic.LoadInt32(&c.BytesReadFromDeadline) > c.ReadThreshold {
|
if c.BytesReadFromDeadline.Load() > c.ReadThreshold {
|
||||||
atomic.StoreInt32(&c.BytesReadFromDeadline, 0)
|
c.BytesReadFromDeadline.Store(0)
|
||||||
// we set both read and write deadlines here otherwise after the request
|
// we set both read and write deadlines here otherwise after the request
|
||||||
// is read writing the response fails with an i/o timeout error
|
// is read writing the response fails with an i/o timeout error
|
||||||
err = c.Conn.SetDeadline(time.Now().Add(c.ReadTimeout))
|
err = c.Conn.SetDeadline(time.Now().Add(c.ReadTimeout))
|
||||||
|
@ -64,13 +62,13 @@ func (c *Conn) Read(b []byte) (n int, err error) {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
n, err = c.Conn.Read(b)
|
n, err = c.Conn.Read(b)
|
||||||
atomic.AddInt32(&c.BytesReadFromDeadline, int32(n))
|
c.BytesReadFromDeadline.Add(int32(n))
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
func (c *Conn) Write(b []byte) (n int, err error) {
|
func (c *Conn) Write(b []byte) (n int, err error) {
|
||||||
if atomic.LoadInt32(&c.BytesWrittenFromDeadline) > c.WriteThreshold {
|
if c.BytesWrittenFromDeadline.Load() > c.WriteThreshold {
|
||||||
atomic.StoreInt32(&c.BytesWrittenFromDeadline, 0)
|
c.BytesWrittenFromDeadline.Store(0)
|
||||||
// we extend the read deadline too, not sure it's necessary,
|
// we extend the read deadline too, not sure it's necessary,
|
||||||
// but it doesn't hurt
|
// but it doesn't hurt
|
||||||
err = c.Conn.SetDeadline(time.Now().Add(c.WriteTimeout))
|
err = c.Conn.SetDeadline(time.Now().Add(c.WriteTimeout))
|
||||||
|
@ -79,7 +77,7 @@ func (c *Conn) Write(b []byte) (n int, err error) {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
n, err = c.Conn.Write(b)
|
n, err = c.Conn.Write(b)
|
||||||
atomic.AddInt32(&c.BytesWrittenFromDeadline, int32(n))
|
c.BytesWrittenFromDeadline.Add(int32(n))
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
32
queue.go
32
queue.go
|
@ -210,7 +210,7 @@ func (p *QueueMetadataPacket) GetData() []byte {
|
||||||
type Queue struct {
|
type Queue struct {
|
||||||
NowPlaying chan *QueueTrackEntry
|
NowPlaying chan *QueueTrackEntry
|
||||||
QueueEmpty chan *QueueTrackEntry
|
QueueEmpty chan *QueueTrackEntry
|
||||||
Duration time.Duration
|
duration atomic.Int64
|
||||||
durationError int64
|
durationError int64
|
||||||
audioQueue *queue.Queue
|
audioQueue *queue.Queue
|
||||||
mounts []*StreamMount
|
mounts []*StreamMount
|
||||||
|
@ -264,6 +264,10 @@ func NewQueue(config *Config) *Queue {
|
||||||
return q
|
return q
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (q *Queue) GetDuration() time.Duration {
|
||||||
|
return time.Duration(q.duration.Load())
|
||||||
|
}
|
||||||
|
|
||||||
func (q *Queue) Wait() {
|
func (q *Queue) Wait() {
|
||||||
q.wg.Wait()
|
q.wg.Wait()
|
||||||
close(q.NowPlaying)
|
close(q.NowPlaying)
|
||||||
|
@ -280,9 +284,9 @@ func (q *Queue) AddTrack(entry *QueueTrackEntry, tail bool) error {
|
||||||
log.Printf("now playing %s\n", e.Path)
|
log.Printf("now playing %s\n", e.Path)
|
||||||
q.NowPlaying <- e
|
q.NowPlaying <- e
|
||||||
for _, mount := range q.mounts {
|
for _, mount := range q.mounts {
|
||||||
mount.MetadataQueue.Enqueue(&QueueMetadataPacket{
|
_ = mount.MetadataQueue.Enqueue(&QueueMetadataPacket{
|
||||||
//TODO: carry sample rate error
|
//TODO: carry sample rate error
|
||||||
sampleNumber: int64(q.Duration * time.Duration(queue.GetSampleRate()) / time.Second),
|
sampleNumber: (q.duration.Load() * int64(queue.GetSampleRate())) / int64(time.Second),
|
||||||
TrackEntry: e,
|
TrackEntry: e,
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
@ -297,7 +301,7 @@ func (q *Queue) AddTrack(entry *QueueTrackEntry, tail bool) error {
|
||||||
|
|
||||||
removeCallback := func(queue *queue.Queue, entry *queue.QueueEntry) {
|
removeCallback := func(queue *queue.Queue, entry *queue.QueueEntry) {
|
||||||
//TODO: carry sample rate error
|
//TODO: carry sample rate error
|
||||||
atomic.AddInt64((*int64)(&q.Duration), int64((time.Second*time.Duration(atomic.LoadUint64(&entry.ReadSamples)))/time.Duration(entry.Source.GetSampleRate())))
|
q.duration.Add(int64((time.Second * time.Duration(entry.ReadSamples.Load())) / time.Duration(entry.Source.GetSampleRate())))
|
||||||
|
|
||||||
q.Remove(entry.Identifier)
|
q.Remove(entry.Identifier)
|
||||||
q.HandleQueue()
|
q.HandleQueue()
|
||||||
|
@ -635,7 +639,7 @@ func (q *Queue) HandleRadioRequest(writer http.ResponseWriter, request *http.Req
|
||||||
writeChannel := make(chan []byte, byteSliceChannelBuffer)
|
writeChannel := make(chan []byte, byteSliceChannelBuffer)
|
||||||
|
|
||||||
requestDone := struct {
|
requestDone := struct {
|
||||||
Done uint32
|
Done atomic.Bool
|
||||||
Lock sync.Mutex
|
Lock sync.Mutex
|
||||||
Error error
|
Error error
|
||||||
}{}
|
}{}
|
||||||
|
@ -649,7 +653,7 @@ func (q *Queue) HandleRadioRequest(writer http.ResponseWriter, request *http.Req
|
||||||
writer.Header().Set("Content-Type", "application/x-audio-packet-stream")
|
writer.Header().Set("Content-Type", "application/x-audio-packet-stream")
|
||||||
|
|
||||||
packetWriteCallback = func(packet packetizer.Packet) error {
|
packetWriteCallback = func(packet packetizer.Packet) error {
|
||||||
if atomic.LoadUint32(&requestDone.Done) == 1 {
|
if requestDone.Done.Load() {
|
||||||
return requestDone.Error
|
return requestDone.Error
|
||||||
}
|
}
|
||||||
if metadataPacket, ok := packet.(*QueueMetadataPacket); ok {
|
if metadataPacket, ok := packet.(*QueueMetadataPacket); ok {
|
||||||
|
@ -661,7 +665,7 @@ func (q *Queue) HandleRadioRequest(writer http.ResponseWriter, request *http.Req
|
||||||
requestDone.Lock.Lock()
|
requestDone.Lock.Lock()
|
||||||
defer requestDone.Lock.Unlock()
|
defer requestDone.Lock.Unlock()
|
||||||
requestDone.Error = errors.New("client ran out of buffer")
|
requestDone.Error = errors.New("client ran out of buffer")
|
||||||
atomic.StoreUint32(&requestDone.Done, 1)
|
requestDone.Done.Store(true)
|
||||||
return requestDone.Error
|
return requestDone.Error
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -678,7 +682,7 @@ func (q *Queue) HandleRadioRequest(writer http.ResponseWriter, request *http.Req
|
||||||
requestDone.Lock.Lock()
|
requestDone.Lock.Lock()
|
||||||
defer requestDone.Lock.Unlock()
|
defer requestDone.Lock.Unlock()
|
||||||
requestDone.Error = errors.New("client ran out of buffer")
|
requestDone.Error = errors.New("client ran out of buffer")
|
||||||
atomic.StoreUint32(&requestDone.Done, 1)
|
requestDone.Done.Store(true)
|
||||||
return requestDone.Error
|
return requestDone.Error
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -697,7 +701,7 @@ func (q *Queue) HandleRadioRequest(writer http.ResponseWriter, request *http.Req
|
||||||
requestDone.Lock.Lock()
|
requestDone.Lock.Lock()
|
||||||
defer requestDone.Lock.Unlock()
|
defer requestDone.Lock.Unlock()
|
||||||
requestDone.Error = errors.New("client ran out of buffer")
|
requestDone.Error = errors.New("client ran out of buffer")
|
||||||
atomic.StoreUint32(&requestDone.Done, 1)
|
requestDone.Done.Store(true)
|
||||||
return requestDone.Error
|
return requestDone.Error
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -776,7 +780,7 @@ func (q *Queue) HandleRadioRequest(writer http.ResponseWriter, request *http.Req
|
||||||
var streamStartOffset int64 = -1
|
var streamStartOffset int64 = -1
|
||||||
|
|
||||||
packetWriteCallback = func(packet packetizer.Packet) error {
|
packetWriteCallback = func(packet packetizer.Packet) error {
|
||||||
if atomic.LoadUint32(&requestDone.Done) == 1 {
|
if requestDone.Done.Load() {
|
||||||
return requestDone.Error
|
return requestDone.Error
|
||||||
}
|
}
|
||||||
if metadataPacket, ok := packet.(*QueueMetadataPacket); ok {
|
if metadataPacket, ok := packet.(*QueueMetadataPacket); ok {
|
||||||
|
@ -826,7 +830,7 @@ func (q *Queue) HandleRadioRequest(writer http.ResponseWriter, request *http.Req
|
||||||
requestDone.Lock.Lock()
|
requestDone.Lock.Lock()
|
||||||
defer requestDone.Lock.Unlock()
|
defer requestDone.Lock.Unlock()
|
||||||
requestDone.Error = errors.New("client ran out of buffer")
|
requestDone.Error = errors.New("client ran out of buffer")
|
||||||
atomic.StoreUint32(&requestDone.Done, 1)
|
requestDone.Done.Store(true)
|
||||||
return requestDone.Error
|
return requestDone.Error
|
||||||
}
|
}
|
||||||
writeChannel <- data
|
writeChannel <- data
|
||||||
|
@ -835,7 +839,7 @@ func (q *Queue) HandleRadioRequest(writer http.ResponseWriter, request *http.Req
|
||||||
} else {
|
} else {
|
||||||
var streamStartOffset int64 = -1
|
var streamStartOffset int64 = -1
|
||||||
packetWriteCallback = func(packet packetizer.Packet) error {
|
packetWriteCallback = func(packet packetizer.Packet) error {
|
||||||
if atomic.LoadUint32(&requestDone.Done) == 1 {
|
if requestDone.Done.Load() {
|
||||||
return requestDone.Error
|
return requestDone.Error
|
||||||
}
|
}
|
||||||
if _, ok := packet.(*QueueMetadataPacket); ok {
|
if _, ok := packet.(*QueueMetadataPacket); ok {
|
||||||
|
@ -846,7 +850,7 @@ func (q *Queue) HandleRadioRequest(writer http.ResponseWriter, request *http.Req
|
||||||
requestDone.Lock.Lock()
|
requestDone.Lock.Lock()
|
||||||
defer requestDone.Lock.Unlock()
|
defer requestDone.Lock.Unlock()
|
||||||
requestDone.Error = errors.New("client ran out of buffer")
|
requestDone.Error = errors.New("client ran out of buffer")
|
||||||
atomic.StoreUint32(&requestDone.Done, 1)
|
requestDone.Done.Store(true)
|
||||||
return requestDone.Error
|
return requestDone.Error
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -882,7 +886,7 @@ func (q *Queue) HandleRadioRequest(writer http.ResponseWriter, request *http.Req
|
||||||
requestDone.Lock.Lock()
|
requestDone.Lock.Lock()
|
||||||
defer requestDone.Lock.Unlock()
|
defer requestDone.Lock.Unlock()
|
||||||
requestDone.Error = errors.New("client ran out of buffer")
|
requestDone.Error = errors.New("client ran out of buffer")
|
||||||
atomic.StoreUint32(&requestDone.Done, 1)
|
requestDone.Done.Store(true)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
//try flush
|
//try flush
|
||||||
|
|
Loading…
Reference in a new issue