added flags to retry errors and stop on errors

cmd/ytsync.go

@@ -12,6 +12,7 @@ import (
 	"strconv"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/lbryio/lbry.go/jsonrpc"
@@ -31,6 +32,8 @@ func init() {
 		Short: "Publish youtube channel into LBRY network.",
 		Run:   ytsync,
 	}
+	ytSyncCmd.Flags().BoolVar(&stopOnError, "stop-on-error", false, "If a video fails, stop publishing")
+	ytSyncCmd.Flags().BoolVar(&retryErrors, "retry-errors", false, "Retry failed publishes")
 	RootCmd.AddCommand(ytSyncCmd)
 }
 
@@ -71,6 +74,8 @@ var (
 	ytAPIKey        string
 	channelID       string
 	lbryChannelName string
+	stopOnError     bool
+	retryErrors     bool
 
 	daemon         *jsonrpc.Client
 	claimAddress   string
@@ -87,6 +92,11 @@ func ytsync(cmd *cobra.Command, args []string) {
 		lbryChannelName = args[2]
 	}
 
+	if stopOnError && retryErrors {
+		log.Errorln("--stop-on-error and --retry-errors are mutually exclusive")
+		return
+	}
+
 	redisPool = &redis.Pool{
 		MaxIdle:     3,
 		IdleTimeout: 5 * time.Minute,
@@ -103,6 +113,15 @@ func ytsync(cmd *cobra.Command, args []string) {
 	var wg sync.WaitGroup
 	videoQueue := make(chan video)
 
+	stopEnqueuing := make(chan struct{})
+	sendStopEnqueuing := sync.Once{}
+
+	var videoErrored atomic.Value
+	videoErrored.Store(false)
+	if stopOnError {
+		log.Println("Will stop publishing if an error is detected")
+	}
+
 	daemon = jsonrpc.NewClient("")
 	videoDirectory, err = ioutil.TempDir("", "ytsync")
 	if err != nil {
@@ -137,15 +156,34 @@ func ytsync(cmd *cobra.Command, args []string) {
 				if !more {
 					return
 				}
-				err := processVideo(v)
-				if err != nil {
-					log.Errorln("error processing video: " + err.Error())
+				if stopOnError && videoErrored.Load().(bool) {
+					log.Println("Video errored. Exiting")
+					return
+				}
+
+				for {
+					err := processVideo(v)
+					if err != nil {
+						log.Errorln("error processing video: " + err.Error())
+						if stopOnError {
+							videoErrored.Store(true)
+							sendStopEnqueuing.Do(func() {
+								stopEnqueuing <- struct{}{}
+							})
+						}
+					}
+					if err != nil && retryErrors {
+						log.Println("Retrying")
+					} else {
+						break
+					}
+
 				}
 			}
 		}()
 	}
 
-	err = enqueueVideosFromChannel(channelID, &videoQueue)
+	err = enqueueVideosFromChannel(channelID, &videoQueue, &stopEnqueuing)
 	if err != nil {
 		panic(err)
 	}
@@ -192,7 +230,7 @@ func ensureChannelOwnership() error {
 	return nil
 }
 
-func enqueueVideosFromChannel(channelID string, videoChan *chan video) error {
+func enqueueVideosFromChannel(channelID string, videoChan *chan video, stopEnqueuing *chan struct{}) error {
 	client := &http.Client{
 		Transport: &transport.APIKey{Key: ytAPIKey},
 	}
@@ -270,14 +308,18 @@ func enqueueVideosFromChannel(channelID string, videoChan *chan video) error {
 	//or sort.Sort(sort.Reverse(byPlaylistPosition(videos)))
 
 	for _, v := range videos {
-		*videoChan <- v
+		select {
+		case *videoChan <- v:
+		case <-*stopEnqueuing:
+			return nil
+		}
 	}
 
 	return nil
 }
 
 func processVideo(v video) error {
-	log.Println("Processing " + v.id)
+	log.Println("Processing " + v.id + " (" + strconv.Itoa(int(v.playlistPosition)) + " in channel)")
 
 	conn := redisPool.Get()
 	defer conn.Close()

jsonrpc/daemon.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"reflect"
+	"sort"
 	"strconv"
 	"strings"
 
@@ -60,6 +61,7 @@ func debugParams(params map[string]interface{}) string {
 		}
 		s = append(s, fmt.Sprintf("%s=%+v", k, v))
 	}
+	sort.Strings(s)
 	return strings.Join(s, " ")
 }
 

jsonrpc/daemon_types.go

@@ -113,6 +113,12 @@ func fixDecodeProto(src, dest reflect.Type, data interface{}) (interface{}, erro
 				return nil, err
 			}
 			return decimal.NewFromFloat(val), nil
+		} else if s, ok := data.(string); ok {
+			d, err := decimal.NewFromString(s)
+			if err != nil {
+				return nil, err
+			}
+			return d, nil
 		}
 
 	case reflect.TypeOf(lbryschema.Metadata_Version(0)):