go/src/v.io/x/sensorlog/slcli/stream.go

// Copyright 2015 The Vanadium Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.

// slcli stream configuration.

package main

import (
	"fmt"
	"io/ioutil"
	"time"

	"v.io/v23/context"
	"v.io/x/lib/cmdline"
	"v.io/x/ref/lib/v23cmd"
	"v.io/x/sensorlog_lite/internal/client"
	"v.io/x/sensorlog_lite/internal/sbmodel"
	"v.io/x/sensorlog_lite/internal/sbutil"
)

var cmdSLStream = &cmdline.Command{
	Name:  "stream",
	Short: "Manage sampling streams",
	Long: `
Create sampling streams.

TODO(ivanpi): Enable/disable, list.
`,
	Children: []*cmdline.Command{cmdSLStreamCreate /*, cmdSLStreamEnable, cmdSLStreamDisable, cmdSLStreamList*/},
}

var cmdSLStreamCreate = &cmdline.Command{
	Runner: v23cmd.RunnerFunc(runSLStreamCreate),
	Name:   "create",
	Short:  "Create new data stream",
	Long: `
Creates a new sampling stream and outputs its identifier.

The sampling script is read from stdin. It must be a bash script that prints
a single floating point result per invocation and exits with a zero status,
otherwise an error is logged instead of the data point. The script will be
run on the specified device at the specified frequency whenever the measuring
daemon is running on the device, as long as the stream is enabled.
`,
	ArgsName: "<device_id> <stream_id> <interval> [<stream_desc>]",
	ArgsLong: `
<device_id> is the identifier of the device to use for sampling.

<stream_id> is the identifier of the stream to be created. It must be unique
            per measuring device.

<interval> is the time interval between two subsequent invocations of the
           measuring script. Measurements may be skipped if the interval is
           too short compared to the device and sampling script speed.

<stream_desc> is a human-readable description of the stream.
              It doesn't need to be unique.
`,
}

func runSLStreamCreate(ctx *context.T, env *cmdline.Env, args []string) error {
	if len(args) < 3 || len(args) > 4 {
		return env.UsageErrorf("expects between 3 and 4 arguments")
	}
	devKey := &sbmodel.KDeviceCfg{
		DevId: args[0],
	}
	streamId := args[1]
	interval, err := time.ParseDuration(args[2])
	if err != nil {
		return fmt.Errorf("failed parsing interval %q: %v", args[2], err)
	}
	desc := ""
	if len(args) > 3 {
		desc = args[3]
	}

	script, err := ioutil.ReadAll(env.Stdin)
	if err != nil {
		return err
	}

	db, err := sbutil.CreateOrOpenDB(ctx, *flagSbService, sbmodel.MasterTables)
	if err != nil {
		return fmt.Errorf("failed opening Syncbase db: %v", err)
	}

	streamKey, err := client.CreateStream(ctx, db, devKey, streamId, string(script), interval, desc)
	if err != nil {
		return err
	}

	fmt.Fprintf(env.Stdout, "%s\n", streamKey.StreamId)

	return nil
}