dwelling-radio/internal/radio/icecast.go

package radio

import (
	"bytes"
	"dwelling-radio/pkg/watcher"
	"encoding/json"
	"io"
	"log"
	"net/http"
	"os"
	"sync"
	"syscall"
	"time"

	"github.com/pkg/errors"
)

const (
	IcecastPlaylistDateFormat = "02/Jan/2006:15:04:05 -0700"
	SongTimeFormat            = "2006 15:04-0700"

	bufferSizePerLine = 512
	// Positions of first and second "|" separator on playlist.log line.
	// The third one may float if amount of listeners will be >= 10.
	separatorOne = 26
	// +1 added to a second one just because it is used only at a start pos of
	// a listeners' field, so there's no need to increment it every time.
	separatorTwo = 38 + 1
)

var (
	currentlyPlaying     Song
	lastPlayedCache      []Song
	lastPlayedCacheMutex sync.Mutex
)

type IcecastStatusDTO struct {
	Icestats struct {
		ServerStartISO8601 string `json:"server_start_iso8601"`
		ServerStartDate    string `json:"server_start"`
		Source             struct {
			Artist       string `json:"artist"`
			Title        string `json:"title"`
			ListenerPeak int    `json:"listener_peak"`
			Listeners    int    `json:"listeners"`
		} `json:"source"`
	} `json:"icestats"`
}

func (is *IcecastStatusDTO) SongName() string {
	return is.Icestats.Source.Artist + " - " + is.Icestats.Source.Title
}

type IcecastStatus struct {
	ListenerPeak int    `json:"listener_peak"`
	Listeners    int    `json:"listeners"`
	SongName     string `json:"song"`
}

func IcecastGetStatus(icecastURL string) (*IcecastStatus, error) {
	resp, err := http.Get(icecastURL)
	if err != nil {
		return &IcecastStatus{SongName: "Offline"}, err
	}

	iceStatDTO := &IcecastStatusDTO{}

	if err := json.NewDecoder(resp.Body).Decode(iceStatDTO); err != nil {
		return &IcecastStatus{}, err
	}

	return &IcecastStatus{
		SongName:     iceStatDTO.SongName(),
		ListenerPeak: iceStatDTO.Icestats.Source.ListenerPeak,
		Listeners:    iceStatDTO.Icestats.Source.Listeners,
	}, nil
}

type Song struct {
	Time      string `json:"time"`
	Listeners string `json:"listeners"`
	Song      string `json:"song"`
}

func IcecastLastSongs(playlistPath string) []Song {
	lastPlayedCacheMutex.Lock()
	defer lastPlayedCacheMutex.Unlock()
	if lpcLen := len(lastPlayedCache); lpcLen > 0 {
		ret := make([]Song, lpcLen)
		copy(ret, lastPlayedCache)
		return ret
	}
	return nil
}

func IcecastLastSong(playlistPath string) *Song {
	lastPlayedCacheMutex.Lock()
	defer lastPlayedCacheMutex.Unlock()
	if lpcLen := len(lastPlayedCache); lpcLen > 0 {
		return &lastPlayedCache[lpcLen-1]
	}
	return nil
}

func icecastCurrentSong(playlistPath string) (*Song, error) {
	fd, err := os.Open(playlistPath)
	if err != nil {
		return nil, err
	}
	defer fd.Close()

	fdSize, _ := fd.Seek(0, io.SeekEnd)
	if fdSize == 0 {
		return nil, nil
	}

	var bufSize int64 = bufferSizePerLine
	if fdSize < bufferSizePerLine {
		bufSize = fdSize
	}

	_, err = fd.Seek(-bufSize, io.SeekEnd)
	if err != nil {
		return nil, err
	}

	buf := make([]byte, bufSize)

	_, err = fd.Read(buf)
	if err != nil {
		return nil, err
	}

	curSongEnd := bytes.LastIndexByte(buf, '\n')
	line := buf[bytes.LastIndexByte(buf[:curSongEnd], '\n')+1 : curSongEnd]

	songTime, _ := time.Parse(IcecastPlaylistDateFormat, string(line[:separatorOne]))

	separatorThree := bytes.LastIndexByte(line, '|')

	return &Song{
		Time:      songTime.Format(SongTimeFormat),
		Listeners: string(line[separatorTwo:separatorThree]),
		Song:      string(line[separatorThree+1:])}, nil
}

type PlaylistLogWatcher struct {
	watcher *watcher.InotifyWatcher
	changed chan uint32
}

func NewPlaylistLogWatcher() *PlaylistLogWatcher {
	return &PlaylistLogWatcher{changed: make(chan uint32)}
}

func (pw *PlaylistLogWatcher) Watch(playlistPath string, n int) (err error) {
	if pw.watcher != nil {
		pw.watcher.Close()
	}

	pw.watcher, err = watcher.NewInotifyWatcher()
	if err != nil {
		return errors.Wrap(err, "cannot instantiate inotify watcher")
	}

	err = pw.watcher.AddWatch(playlistPath, watcher.ModIgnMask)
	if err != nil {
		return errors.Wrap(err, "cannot set a playlist to watch")
	}

	pw.watcher.WatchForMask(pw.changed, watcher.ModIgnMask)

	if lastPlayedCache == nil {
		lastPlayedCache = make([]Song, 0, n)
	}

	cur, err := icecastCurrentSong(playlistPath)
	if err != nil {
		log.Println("failed to fetch current song:", err)
	} else if cur != nil && currentlyPlaying.Time != "" {
		currentlyPlaying = *cur
	}

	go func() {
		for {
			mask := <-pw.changed

			if mask&syscall.IN_MODIFY > 0 {
				lastPlayedCacheMutex.Lock()
				song, err := icecastCurrentSong(playlistPath)
				if err == nil && song != nil {
					CheckAndUpdateMostListenedSong(song, &currentlyPlaying)
					if currentlyPlaying.Time == "" {
						currentlyPlaying = *song
					} else {
						currentlyPlaying.Listeners = song.Listeners
						if len(lastPlayedCache) == n {
							lastPlayedCache = append(lastPlayedCache[1:], currentlyPlaying)
						} else {
							lastPlayedCache = append(lastPlayedCache, currentlyPlaying)
						}
						currentlyPlaying = *song
					}
				} else if err != nil {
					log.Println("failed to retrieve last songs:", err)
				}
				lastPlayedCacheMutex.Unlock()
			} else if mask&syscall.IN_IGNORED > 0 {
				pw.Close()
				pw.Watch(playlistPath, n)
				return
			}
		}
	}()

	return nil
}

func (pw *PlaylistLogWatcher) Close() {
	pw.watcher.Close()
}