added converter fetcher

2 files changed, 148 insertions, 12 deletions

diff --git a/rhimport/converter.go b/rhimport/converter.go
new file mode 100644
index 0000000..4b68faa
--- /dev/null
+++ b/rhimport/converter.go
@@ -0,0 +1,115 @@
+//
+//  rhimportd
+//
+//  The Radio Helsinki Rivendell Import Daemon
+//
+//
+//  Copyright (C) 2015-2016 Christian Pointner <equinox@helsinki.at>
+//
+//  This file is part of rhimportd.
+//
+//  rhimportd 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.
+//
+//  rhimportd 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 rhimportd. If not, see <http://www.gnu.org/licenses/>.
+//
+
+package rhimport
+
+import (
+	"io"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strings"
+)
+
+type Converter interface {
+	io.WriteCloser
+	GetResult() (result string, err error)
+}
+
+type ConverterResult struct {
+	output string
+	err    error
+}
+
+//
+// NUll Converter aka File Writer
+//
+
+type NullConverter struct {
+	file *os.File
+}
+
+func NewNullConverter(filename string) (n *NullConverter, newFilename string, err error) {
+	n = &NullConverter{}
+	rhl.Printf("null-converter: opening file '%s'", filename)
+	newFilename = filename
+	n.file, err = os.OpenFile(filename, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0600)
+	return
+}
+
+func (c *NullConverter) Write(p []byte) (n int, err error) {
+	return c.file.Write(p)
+}
+
+func (c *NullConverter) Close() (err error) {
+	return c.file.Close()
+}
+
+func (c *NullConverter) GetResult() (result string, err error) {
+	return "", nil
+}
+
+//
+// FFMpeg Converter: converts all files into flac
+//
+
+type FFMpegConverter struct {
+	cmd    *exec.Cmd
+	pipe   io.WriteCloser
+	result chan ConverterResult
+}
+
+func NewFFMpegConverter(filename string) (ff *FFMpegConverter, filenameFlac string, err error) {
+	ff = &FFMpegConverter{}
+	ext := filepath.Ext(filename)
+	filenameFlac = strings.TrimSuffix(filename, ext) + ".flac"
+	rhl.Printf("ffmpeg-converter: starting ffmpeg for file '%s' (had extension: '%s')", filenameFlac, ext)
+	ff.cmd = exec.Command("ffmpeg", "-loglevel", "warning", "-i", "-", "-map_metadata", "0", "-f", "flac", filenameFlac)
+	if ff.pipe, err = ff.cmd.StdinPipe(); err != nil {
+		return nil, "", err
+	}
+
+	ff.result = make(chan ConverterResult, 1)
+	go func() {
+		output, err := ff.cmd.CombinedOutput()
+		ff.result <- ConverterResult{strings.TrimSpace(string(output)), err}
+	}()
+	return
+}
+
+func (ff *FFMpegConverter) Write(p []byte) (n int, err error) {
+	return ff.pipe.Write(p)
+}
+
+func (ff *FFMpegConverter) Close() (err error) {
+	return ff.pipe.Close()
+}
+
+func (ff *FFMpegConverter) GetResult() (result string, err error) {
+	if ff.result != nil {
+		r := <-ff.result
+		return r.output, r.err
+	}
+	return "", nil
+}
diff --git a/rhimport/fetcher.go b/rhimport/fetcher.go
index 5fa442e..fde9da8 100644
--- a/rhimport/fetcher.go
+++ b/rhimport/fetcher.go
@@ -47,13 +47,13 @@ type FetcherCurlCBData struct {
 	basepath   string
 	filename   string
 	remotename string
-	file       *os.File
+	conv       Converter
 	writeError error
 }
 
 func (self *FetcherCurlCBData) Cleanup() {
-	if self.file != nil {
-		self.file.Close()
+	if self.conv != nil {
+		self.conv.Close()
 	}
 }
 
@@ -73,7 +73,7 @@ func curlHeaderCallback(ptr []byte, userdata interface{}) bool {
 
 func curlWriteCallback(ptr []byte, userdata interface{}) bool {
 	data := userdata.(*FetcherCurlCBData)
-	if data.file == nil {
+	if data.conv == nil {
 		if data.filename == "" {
 			name := path.Clean("/" + data.remotename)
 			if name == "/" {
@@ -82,15 +82,16 @@ func curlWriteCallback(ptr []byte, userdata interface{}) bool {
 			}
 			data.filename = filepath.Join(data.basepath, name)
 		}
-		fp, err := os.OpenFile(data.filename, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0600)
+		nc, newFilename, err := NewFFMpegConverter(data.filename)
 		if err != nil {
-			rhl.Printf("Unable to create file %s: %s", data.filename, err)
+			rhl.Printf("Unable to create converter for file %s: %s", data.filename, err)
 			data.writeError = err
 			return false
 		}
-		data.file = fp
+		data.filename = newFilename
+		data.conv = nc
 	}
-	if _, err := data.file.Write(ptr); err != nil {
+	if _, err := data.conv.Write(ptr); err != nil {
 		rhl.Printf("Unable to write file %s: %s", data.filename, err)
 		data.writeError = err
 		return false
@@ -155,8 +156,12 @@ func fetchFileCurl(ctx *Context, res *Result, uri *url.URL) (err error) {
 	easy.Setopt(curl.OPT_PROGRESSFUNCTION, curlProgressCallback)
 	easy.Setopt(curl.OPT_PROGRESSDATA, cbdata)
 
-	if err = easy.Perform(); err != nil {
-		if cbdata.file != nil {
+	err = easy.Perform()
+	cbdata.conv.Close()
+	rhl.Printf("waiting for converter to finish...")
+	convOut, convErr := cbdata.conv.GetResult()
+	if err != nil || cbdata.writeError != nil || convErr != nil {
+		if cbdata.conv != nil {
 			rhdl.Printf("Removing stale file: %s", cbdata.filename)
 			os.Remove(cbdata.filename)
 			os.Remove(path.Dir(cbdata.filename))
@@ -168,6 +173,12 @@ func fetchFileCurl(ctx *Context, res *Result, uri *url.URL) (err error) {
 		if cbdata.writeError != nil {
 			err = cbdata.writeError
 		}
+		if convErr != nil {
+			err = convErr
+			if convOut != "" {
+				rhl.Printf("converter output: %s", convOut)
+			}
+		}
 		err = fmt.Errorf("curl-fetcher('%s'): %s", ctx.SourceUri, err)
 		rhl.Println(err)
 		return
@@ -262,8 +273,12 @@ func fetchFileArchiv(ctx *Context, res *Result, uri *url.URL) (err error) {
 	easy.Setopt(curl.OPT_PROGRESSDATA, cbdata)
 
 	rhdl.Printf("importing archiv file from %s", scpuri)
-	if err = easy.Perform(); err != nil {
-		if cbdata.file != nil {
+	err = easy.Perform()
+	cbdata.conv.Close()
+	rhl.Printf("waiting for converter to finish...")
+	convOut, convErr := cbdata.conv.GetResult()
+	if err != nil || cbdata.writeError != nil || convErr != nil {
+		if cbdata.conv != nil {
 			rhdl.Printf("Removing stale file: %s", cbdata.filename)
 			os.Remove(cbdata.filename)
 			os.Remove(path.Dir(cbdata.filename))
@@ -275,6 +290,12 @@ func fetchFileArchiv(ctx *Context, res *Result, uri *url.URL) (err error) {
 		if cbdata.writeError != nil {
 			err = cbdata.writeError
 		}
+		if convErr != nil {
+			err = convErr
+			if convOut != "" {
+				rhl.Printf("converter output: %s", convOut)
+			}
+		}
 		err = fmt.Errorf("archiv-fetcher('%s'): %s", ctx.SourceUri, err)
 		rhl.Println(err)
 		return