stream

var Command = &cobra.Command{
	Use:   "stream <stream>",
	Short: "Stream a feed to stdout",
	Long:  "Stream newline-delimited JSON from proxies, anonymizers, torrents, and Helios honeypot feeds.",
	Args:  cobra.ExactArgs(1),
	Run: func(cmd *cobra.Command, args []string) {
		stream, ok := feed.Find(args[0])
		if !ok {
			timber.FatalMsgf("unknown stream %q; valid streams: %s", args[0], feed.Names())
		}
		validateFilters()
		if flags.duration < 0 {
			timber.FatalMsgf("duration must be positive: %s", flags.duration)
		}

		config, err := conf.Read()
		if err != nil {
			app.Fatal(err, "failed to read configuration file")
		}

		client, err := auth.SynthientClient(config)
		if err != nil {
			app.Fatal(err, "failed to create synthient client")
		}
		config.ApplyToClient(&client)

		if !flags.noPreflight {
			err = access.Require(client, access.Required(stream, "stream"))
			if err != nil {
				app.Fatal(err, "failed stream access preflight")
			}
		}

		out, closeOut := output.Open(flags.output)
		defer closeOut()

		ctx := context.Background()
		if flags.duration > 0 {
			var cancel context.CancelFunc
			ctx, cancel = context.WithTimeout(ctx, flags.duration)
			defer cancel()
		}

		if !flags.reconnect {
			seq, buildErr := buildSeq(ctx, &client, stream)
			if buildErr != nil {
				app.Fatal(buildErr, "failed to initialize stream")
			}
			result := streamOnce(ctx, seq, out, flags.maxEvents)
			if result.Err != nil {
				app.Fatal(result.Err, "failed to stream feed")
			}
			return
		}

		attempt := 0
		total := 0
		for {
			remaining := 0
			if flags.maxEvents > 0 {
				remaining = flags.maxEvents - total
				if remaining <= 0 {
					return
				}
			}
			seq, buildErr := buildSeq(ctx, &client, stream)
			if buildErr != nil {
				app.Fatal(buildErr, "failed to initialize stream")
			}
			result := streamOnce(ctx, seq, out, remaining)
			total += result.Count
			if result.Done {
				return
			}
			err = result.Err
			if err == nil {
				attempt = 0
				continue
			}
			if errors.Is(err, synthient.ErrBadRequest) ||
				errors.Is(err, synthient.ErrUnauthorized) ||
				errors.Is(err, synthient.ErrPaymentRequired) ||
				errors.Is(err, synthient.ErrNoToken) {
				app.Fatal(err, "failed to stream feed")
			}
			delay := retryDelay(attempt)
			timber.Warningf("stream %s disconnected, reconnecting in %s", stream.Name, delay)
			select {
			case <-time.After(delay):
			case <-ctx.Done():
				return
			}
			attempt++
		}
	},
}