path: root/player/player.go

//
//  rhlibrary
//
//  The Radio Helsinki Rivendell Library
//
//
//  Copyright (C) 2016 Christian Pointner <equinox@helsinki.at>
//
//  This file is part of rhlibrary.
//
//  rhlibrary is free software: you can redistribute it and/or modify
//  it under the terms of the GNU General Public License as published by
//  the Free Software Foundation, either version 3 of the License, or
//  any later version.
//
//  rhlibrary is distributed in the hope that it will be useful,
//  but WITHOUT ANY WARRANTY; without even the implied warranty of
//  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
//  GNU General Public License for more details.
//
//  You should have received a copy of the GNU General Public License
//  along with rhlibrary. If not, see <http://www.gnu.org/licenses/>.
//

package player

import (
	"fmt"
	"io/ioutil"
	"log"
	"os"
	"path"
	"time"

	"github.com/spreadspace/go-gstreamer"
)

func init() {
	gst.Init(nil)
}

type loadResult struct {
	err error
}

type loadRequest struct {
	cart     uint
	cut      uint
	response chan<- loadResult
}

type playResult struct {
	err error
}

type playRequest struct {
	response chan<- playResult
}

type pauseResult struct {
	err error
}

type pauseRequest struct {
	response chan<- pauseResult
}

type stopResult struct {
	err error
}

type stopRequest struct {
	response chan<- stopResult
}

type addUpdateHandlerResult struct {
	err error
}

type addUpdateHandlerRequest struct {
	userdata interface{}
	callback UpdateCB
	response chan<- addUpdateHandlerResult
}

type MeterChannel struct {
	Peak  float64
	Decay float64
}
type Meter []MeterChannel
type updateData struct {
	len   time.Duration
	pos   time.Duration
	meter Meter
}
type UpdateCB func(len time.Duration, pos time.Duration, meter Meter, userdata interface{}) bool
type pUpdateCB struct {
	cb       UpdateCB
	userdata interface{}
}

type State int

const (
	IDLE State = 1 + iota
	PAUSED
	PLAYING
)

type Player struct {
	pipe                 *gst.Pipeline
	src                  *gst.Element
	level                *gst.Element
	bus                  *gst.Bus
	basepath             string
	stdlog               *log.Logger
	dbglog               *log.Logger
	state                State
	len                  time.Duration
	loadChan             chan loadRequest
	playChan             chan playRequest
	pauseChan            chan pauseRequest
	stopChan             chan stopRequest
	updateChan           chan updateData
	addUpdateHandlerChan chan addUpdateHandlerRequest
	updateCBs            []*pUpdateCB
}

func (p *Player) onMessage(bus *gst.Bus, msg *gst.Message) {
	switch msg.GetType() {
	case gst.MESSAGE_EOS:
		p.pipe.SetState(gst.STATE_NULL)
		p.state = IDLE
		p.len = 0
		p.updateChan <- updateData{len: p.len}
	case gst.MESSAGE_WARNING:
		warn, _ := msg.ParseWarning()
		p.stdlog.Printf("GStreamer Pipeline Warning: %s", warn)
	case gst.MESSAGE_ERROR:
		p.pipe.SetState(gst.STATE_NULL)
		p.state = IDLE
		p.len = 0
		p.updateChan <- updateData{len: p.len}
		err, _ := msg.ParseError()
		p.stdlog.Printf("GStreamer Pipeline Error: %s", err)
	case gst.MESSAGE_ASYNC_DONE:
		len, ok := p.pipe.QueryDuration(gst.FORMAT_TIME)
		if ok {
			p.len = time.Duration(len)
			p.updateChan <- updateData{len: p.len}
		} else {
			p.stdlog.Printf("GStreamer Pipeline Error: unable to query duration of file")
		}
	case gst.MESSAGE_ELEMENT:
		s := msg.GetStructure()
		peak := s.Data["peak"].(gst.GValueArray)
		decay := s.Data["decay"].(gst.GValueArray)

		meter := make([]MeterChannel, len(peak))
		for i := 0; i < len(peak); i++ {
			meter[i] = MeterChannel{Peak: peak[i].(float64), Decay: decay[i].(float64)}
		}
		pos := time.Duration(s.Data["stream-time"].(uint64))
		p.updateChan <- updateData{len: p.len, pos: pos, meter: meter}
	case gst.MESSAGE_STATE_CHANGED:
	default:
		p.dbglog.Printf("GStreamer Message: unknown type '%s'", msg.GetTypeName())
	}
}

func (p *Player) load(cart, cut uint) (resp loadResult) {
	filename := path.Join(p.basepath, fmt.Sprintf("%06d_%03d.wav", cart, cut))

	var file *os.File
	if file, resp.err = os.Open(filename); resp.err != nil {
		resp.err = fmt.Errorf("player: %s", resp.err)
		return
	}
	if info, err := file.Stat(); err != nil {
		resp.err = fmt.Errorf("player: %s", err)
		return
	} else {
		if info.IsDir() {
			resp.err = fmt.Errorf("player error: '%s' is a directory", filename)
			return
		}
	}
	file.Close()

	if p.state != IDLE {
		p.pipe.SetState(gst.STATE_NULL)
		p.state = IDLE
	}
	p.src.SetProperty("uri", "file://"+filename)
	p.pipe.SetState(gst.STATE_PAUSED)
	p.state = PAUSED
	return
}

func (p *Player) play() (resp playResult) {
	if p.state != PAUSED {
		resp.err = fmt.Errorf("player: no file loaded")
		return
	}
	p.pipe.SetState(gst.STATE_PLAYING)
	p.state = PLAYING
	return
}

func (p *Player) pause() (resp pauseResult) {
	if p.state != PLAYING {
		resp.err = fmt.Errorf("player: not playing")
		return
	}
	p.pipe.SetState(gst.STATE_PAUSED)
	p.state = PAUSED
	return
}

func (p *Player) stop() (resp stopResult) {
	switch p.state {
	case IDLE:
		p.pipe.SetState(gst.STATE_NULL)
		p.state = IDLE
	case PLAYING:
		fallthrough
	case PAUSED:
		p.pipe.SetState(gst.STATE_NULL)
		p.pipe.SetState(gst.STATE_PAUSED)
		p.state = PAUSED
	}
	return
}

func (p *Player) addUpdateHandler(callback UpdateCB, userdata interface{}) (resp addUpdateHandlerResult) {
	p.updateCBs = append(p.updateCBs, &pUpdateCB{callback, userdata})
	return
}

func (p *Player) sendUpdate(len time.Duration, pos time.Duration, meter Meter) {
	for _, cb := range p.updateCBs {
		if cb.cb != nil {
			if keep := cb.cb(len, pos, meter, cb.userdata); !keep {
				cb.cb = nil
			}
		}
	}
	return
}

func (p *Player) dispatchRequests() {
	p.bus.AddSignalWatch()
	p.bus.Connect("message", func(bus *gst.Bus, msg *gst.Message) { p.onMessage(bus, msg) })

	for {
		select {
		case req := <-p.loadChan:
			req.response <- p.load(req.cart, req.cut)
		case req := <-p.playChan:
			req.response <- p.play()
		case req := <-p.pauseChan:
			req.response <- p.pause()
		case req := <-p.stopChan:
			req.response <- p.stop()
		case req := <-p.addUpdateHandlerChan:
			req.response <- p.addUpdateHandler(req.callback, req.userdata)
		case u := <-p.updateChan:
			p.sendUpdate(u.len, u.pos, u.meter)
		}
	}
}

func (p *Player) createPipeline() (err error) {
	if p.pipe, err = gst.PipelineNew("rhlibrary"); err != nil {
		return
	}

	var conv1, conv2, sink *gst.Element
	if p.src, err = gst.ElementFactoryMake("uridecodebin", "source"); err != nil {
		return
	}
	if conv1, err = gst.ElementFactoryMake("audioconvert", "conv1"); err != nil {
		return
	}
	if p.level, err = gst.ElementFactoryMake("level", "meter"); err != nil {
		return
	}
	p.level.SetProperty("message", true)
	p.level.SetProperty("interval", 25000000)
	p.level.SetProperty("peak-ttl", 300000000)
	p.level.SetProperty("peak-falloff", 25)
	if conv2, err = gst.ElementFactoryMake("audioconvert", "conv2"); err != nil {
		return
	}
	if sink, err = gst.ElementFactoryMake("autoaudiosink", "sink"); err != nil {
		return
	}

	p.pipe.Add(p.src)
	p.pipe.Add(conv1)
	p.pipe.Add(p.level)
	p.pipe.Add(conv2)
	p.pipe.Add(sink)

	sinkpad, err := conv1.GetStaticPad("sink")
	if err != nil {
		return fmt.Errorf("player error getting sink pad from conv1: %s", err)
	}
	p.src.Connect("pad-added", func(_ interface{}, srcpad *gst.Pad) {
		if ret := srcpad.Link(sinkpad); ret != gst.PAD_LINK_OK {
			p.stdlog.Println("player error linking src with conv1 (code: %v)", ret)
		}
		p.dbglog.Println("player succesfully linked src with conv1")
	})
	conv1.Link(p.level)
	p.level.Link(conv2)
	conv2.Link(sink)

	if p.bus, err = p.pipe.GetBus(); err != nil {
		return
	}
	return
}

// *********************************************************
// Public Interface

type PlayerChan struct {
	load             chan<- loadRequest
	play             chan<- playRequest
	pause            chan<- pauseRequest
	stop             chan<- stopRequest
	addUpdateHandler chan addUpdateHandlerRequest
}

func (p *PlayerChan) Load(cart, cut uint) error {
	resCh := make(chan loadResult)
	req := loadRequest{}
	req.cart = cart
	req.cut = cut
	req.response = resCh
	p.load <- req

	res := <-resCh
	if res.err != nil {
		return res.err
	}
	return nil
}

func (p *PlayerChan) Play() error {
	resCh := make(chan playResult)
	req := playRequest{}
	req.response = resCh
	p.play <- req

	res := <-resCh
	if res.err != nil {
		return res.err
	}
	return nil
}

func (p *PlayerChan) Pause() error {
	resCh := make(chan pauseResult)
	req := pauseRequest{}
	req.response = resCh
	p.pause <- req

	res := <-resCh
	if res.err != nil {
		return res.err
	}
	return nil
}

func (p *PlayerChan) Stop() error {
	resCh := make(chan stopResult)
	req := stopRequest{}
	req.response = resCh
	p.stop <- req

	res := <-resCh
	if res.err != nil {
		return res.err
	}
	return nil
}

func (p *PlayerChan) AddUpdateHandler(callback UpdateCB, userdata interface{}) error {
	resCh := make(chan addUpdateHandlerResult)
	req := addUpdateHandlerRequest{}
	req.callback = callback
	req.userdata = userdata
	req.response = resCh
	p.addUpdateHandler <- req

	res := <-resCh
	if res.err != nil {
		return res.err
	}
	return nil
}

func (p *Player) GetInterface() *PlayerChan {
	ch := &PlayerChan{}
	ch.load = p.loadChan
	ch.play = p.playChan
	ch.pause = p.pauseChan
	ch.stop = p.stopChan
	ch.addUpdateHandler = p.addUpdateHandlerChan
	return ch
}

func NewPlayer(basepath string, stdlog *log.Logger, dbglog *log.Logger) (p *Player, err error) {
	p = &Player{}
	p.basepath = path.Clean(basepath)
	p.len = 0
	if stdlog != nil {
		p.stdlog = stdlog
	} else {
		p.stdlog = log.New(ioutil.Discard, "autoplayrhrd-go.player", log.LstdFlags)
	}
	if dbglog != nil {
		p.dbglog = dbglog
	} else {
		p.dbglog = log.New(ioutil.Discard, "rhrd-go.player-dbg", log.LstdFlags)
	}
	p.loadChan = make(chan loadRequest)
	p.playChan = make(chan playRequest)
	p.pauseChan = make(chan pauseRequest)
	p.stopChan = make(chan stopRequest)
	p.updateChan = make(chan updateData, 20)
	p.addUpdateHandlerChan = make(chan addUpdateHandlerRequest)

	if err = p.createPipeline(); err != nil {
		return
	}

	go p.dispatchRequests()
	return
}