func (g *importCmd) Run() {
	data, err := ioutil.ReadFile(flag.Arg(1))
	flow.Check(err)

	var values map[string]map[string]interface{}
	err = json.Unmarshal(data, &values)
	flow.Check(err)

	OpenDatabase()
	for prefix, entries := range values {
		var ndel, nadd int

		// get and print all the key/value pairs from the database
		dbIterateOverKeys(prefix, "", func(k string, v []byte) {
			err = db.Delete([]byte(k), nil)
			flow.Check(err)
			ndel++
		})

		for k, v := range entries {
			val, err := json.Marshal(v)
			flow.Check(err)
			err = db.Put([]byte(prefix+k), val, nil)
			flow.Check(err)
			nadd++
		}

		fmt.Printf("%d deleted, %d added for prefix %q\n", ndel, nadd, prefix)
	}
}

func (w *Logger) logOneLine(asof time.Time, text, port string) {
	// figure out name of logfile based on UTC date, with daily rotation
	year, month, day := asof.Date()
	path := fmt.Sprintf("%s/%d", w.dir, year)
	err := os.MkdirAll(path, os.ModePerm)
	flow.Check(err)
	// e.g. "./logger/2014/20140122.txt"
	datePath := fmt.Sprintf("%s/%d.txt", path, (year*100+int(month))*100+day)

	if w.fd == nil || datePath != w.fd.Name() {
		if w.fd != nil {
			name := w.fd.Name()
			w.fd.Close()
			w.Out.Send(name) // report the closed file
		}
		mode := os.O_WRONLY | os.O_APPEND | os.O_CREATE
		fd, err := os.OpenFile(datePath, mode, os.ModePerm)
		flow.Check(err)
		w.fd = fd
	}
	// append a new log entry, here is an example of the format used:
	// 	L 01:02:03.537 usb-A40117UK OK 9 25 54 66 235 61 210 226 33 19
	hour, min, sec := asof.Clock()
	line := fmt.Sprintf("L %02d:%02d:%02d.%03d %s %s\n",
		hour, min, sec, jeebus.TimeToMs(asof)%1000, port, text)
	w.fd.WriteString(line)
}

// Start collecting values, emit aggregated results when a time step occurs.
func (g *Aggregator) Run() {
	g.stats = map[string]accumulator{}

	g.step = "1h"
	if s, ok := <-g.Step; ok {
		g.step = s.(string)
	}
	d, err := time.ParseDuration(g.step)
	flow.Check(err)
	g.stepMs = d.Nanoseconds() / 1e6

	// collect data and aggregate for each parameter
	for m := range g.In {
		if t, ok := m.(flow.Tag); ok {
			n := strings.LastIndex(t.Tag, "/")
			// expects input tags like these:
			// 	sensor/meterkast/c3/1396556362024 = 2396
			if n > 0 {
				prefix := t.Tag[:n+1]
				ms, err := strconv.ParseInt(t.Tag[n+1:], 10, 64)
				flow.Check(err)
				g.process(prefix, ms, t.Msg.(int))
			}
		}
	}

	for k := range g.stats {
		g.flush(k)
	}
}

// Start listening and subscribing to MQTT.
func (w *MQTTSub) Run() {
	port := getInputOrConfig(w.Port, "MQTT_PORT")
	sock, err := net.Dial("tcp", port)
	flow.Check(err)
	client := mqtt.NewClientConn(sock)
	err = client.Connect("", "")
	flow.Check(err)

	for t := range w.In {
		topic := t.(string)
		glog.V(2).Infoln("mqtt-sub", topic)
		if strings.HasSuffix(topic, "/") {
			topic += "#" // if it looks like a prefix, append "#" for MQTT
		}
		client.Subscribe([]proto.TopicQos{{
			Topic: topic,
			Qos:   proto.QosAtMostOnce,
		}})
	}

	for m := range client.Incoming {
		payload := []byte(m.Payload.(proto.BytesPayload))
		// try to decode as JSON, but leave as []byte if that fails
		var any interface{}
		if err = json.Unmarshal(payload, &any); err != nil {
			any = payload
		}
		w.Out.Send(flow.Tag{m.TopicName, any})
	}
}

func dbGet(key string) (any interface{}) {
	glog.V(3).Infoln("get", key)
	data, err := db.Get([]byte(key), nil)
	if err == leveldb.ErrNotFound {
		return nil
	}
	flow.Check(err)
	err = json.Unmarshal(data, &any)
	flow.Check(err)
	return
}

// Parse description into a map of strings, set the "$" entry to a []byte value.
func ParseDescription(desc string) Map {
	// expects mime-type header followed by optional empty line and description
	b := bufio.NewReader(bytes.NewBufferString(desc + "\n\n"))
	header, err := textproto.NewReader(b).ReadMIMEHeader()
	flow.Check(err)
	t, err := ioutil.ReadAll(b)
	flow.Check(err)
	result := Map{"$": bytes.TrimSpace(t)}
	for k, v := range header {
		result[k] = strings.Join(v, "\n")
	}
	return result
}

// Start reading filenames and emit a <file> tag followed by the decoded JSON.
func (w *ReadFileJSON) Run() {
	for m := range w.In {
		if name, ok := m.(string); ok {
			data, err := ioutil.ReadFile(name)
			flow.Check(err)
			w.Out.Send(flow.Tag{"<file>", name})
			var any interface{}
			err = json.Unmarshal(data, &any)
			flow.Check(err)
			m = any
		}
		w.Out.Send(m)
	}
}

// Start decoding smaRelay packets.
func (w *SmaRelay) Run() {
	var vec, prev [7]uint16
	for m := range w.In {
		if v, ok := m.([]byte); ok && len(v) >= 12 {
			buf := bytes.NewBuffer(v[1:])
			err := binary.Read(buf, binary.LittleEndian, &vec)
			flow.Check(err)
			result := map[string]int{
				"<reading>": 1,
				"acw":       int(vec[2]),
				"dcv1":      int(vec[3]),
				"dcv2":      int(vec[4]),
			}
			if vec[0] != prev[0] {
				result["yield"] = int(vec[0])
			}
			if vec[1] != prev[1] {
				result["total"] = int(vec[1])
			}
			if vec[2] != 0 {
				result["dcw1"] = int(vec[5])
				result["dcw2"] = int(vec[6])
			}
			copy(prev[:], vec[:])
			m = result
		}

		w.Out.Send(m)
	}
}

// Start decoding homePower packets.
func (w *HomePower) Run() {
	var vec, prev [6]uint16
	for m := range w.In {
		if v, ok := m.([]byte); ok && len(v) >= 12 {
			buf := bytes.NewBuffer(v[1:])
			err := binary.Read(buf, binary.LittleEndian, &vec)
			flow.Check(err)
			result := map[string]int{"<reading>": 1}
			if vec[0] != prev[0] {
				result["c1"] = int(vec[0])
				result["p1"] = time2watt(int(vec[1]))
			}
			if vec[2] != prev[2] {
				result["c2"] = int(vec[2])
				result["p2"] = time2watt(int(vec[3]))
			}
			if vec[4] != prev[4] {
				result["c3"] = int(vec[4])
				result["p3"] = time2watt(int(vec[5]))
			}
			copy(prev[:], vec[:])
			if len(result) == 1 {
				continue
			}
			m = result
		}

		w.Out.Send(m)
	}
}

// Start running the JavaScript engine.
func (w *JavaScript) Run() {
	if cmd, ok := <-w.Cmd; ok {
		// initial setup
		engine := otto.New()

		// define a callback for sending messages to Out
		engine.Set("emitOut", func(call otto.FunctionCall) otto.Value {
			out, err := call.Argument(0).Export()
			flow.Check(err)
			w.Out.Send(out)
			return otto.UndefinedValue()
		})

		// process the command input
		if _, err := engine.Run(cmd.(string)); err != nil {
			glog.Fatal(err)
		}

		// only start the processing loop if the "onIn" handler exists
		value, err := engine.Get("onIn")
		if err == nil && value.IsFunction() {
			for in := range w.In {
				engine.Call("onIn", nil, in)
			}
		}
	}
}

// Start looking up node ID's in the node map.
func (w *NodeMap) Run() {

	defaultBand := 868 //TODO:Change this to input parameter

	nodeMap := map[NodeMapKey]string{}
	locations := map[NodeMapKey]string{}
	for m := range w.Info {
		f := strings.Split(m.(string), ",")

		key := NodeMapKey{}
		if ok, err := key.Unmarshal(f[0]); !ok {
			flow.Check(err)
		}

		//for the case where the default data has not been changed as in:
		// { data: "RFg5i2,roomNode,boekenkast JC",  to: "nm.Info" }
		//this will automatically be incorporated into the defaultBand network.
		if key.band == 0 {
			key.band = defaultBand
		}

		nodeMap[key] = f[1]

		if len(f) > 2 {
			locations[key] = f[2]
		}
	}

	key := NodeMapKey{}
	for m := range w.In {
		w.Out.Send(m)

		if data, ok := m.(map[string]int); ok {

			switch {
			case data["<RF12demo>"] > 0:
				key.group = data["group"]
				key.band = data["band"]
			case data["<node>"] > 0:
				key.node = data["<node>"]
				if loc, ok := locations[key]; ok {
					w.Out.Send(flow.Tag{"<location>", loc})
				} else {
					w.Missing.Send(key)
					//fmt.Printf("Location NOT found:%+v", key)
				}
				if tag, ok := nodeMap[key]; ok {
					w.Out.Send(flow.Tag{"<dispatch>", tag})
				} else {
					//fmt.Printf("NodeMap NOT found:%+v", key)
					w.Missing.Send(key)
					w.Out.Send(flow.Tag{"<dispatch>", ""})
				}
			}
		}
	}
}

func dbPut(key string, value interface{}) {
	glog.V(2).Infoln("put", key, value)
	if value != nil {
		data, err := json.Marshal(value)
		flow.Check(err)
		db.Put([]byte(key), data, nil)
	} else {
		db.Delete([]byte(key), nil)
	}
}

func (w *FbpParser) parseFbp(lines []string) {
	if len(lines) > 0 {
		fbp := &Fbp{Buffer: strings.Join(lines, "\n")}
		fbp.Init()
		err := fbp.Parse()
		flow.Check(err)
		// fbp.Execute()
		w.Out.Send(true)
		// fbp.PrintSyntaxTree()
	}
}

func openDatabase() {
	// opening the database takes time, make sure we don't re-enter this code
	once.Do(func() {
		dbPath := flow.Config["DATA_DIR"]
		if dbPath == "" {
			glog.Fatalln("cannot open database, DATA_DIR not set")
		}
		ldb, err := leveldb.OpenFile(dbPath, nil)
		flow.Check(err)
		db = ldb
	})
}

func (w *wsHead) Run() {
	for {
		var msg interface{}
		err := websocket.JSON.Receive(w.ws, &msg)
		if err == io.EOF {
			break
		}
		flow.Check(err)
		if s, ok := msg.(string); ok {
			id := w.ws.Request().Header.Get("Sec-Websocket-Key")
			fmt.Println("msg <"+id[:4]+">:", s)
		} else {
			w.Out.Send(msg)
		}
	}
}

// Open the database and start listening to incoming get/put/keys requests.
func (w *LevelDB) Run() {
	openDatabase()
	for m := range w.In {
		if tag, ok := m.(flow.Tag); ok {
			switch tag.Tag {
			case "<keys>":
				w.Out.Send(m)
				for _, s := range dbKeys(tag.Msg.(string)) {
					w.Out.Send(s)
				}
			case "<get>":
				w.Out.Send(m)
				w.Out.Send(dbGet(tag.Msg.(string)))
			case "<clear>":
				prefix := tag.Msg.(string)
				glog.V(2).Infoln("clear", prefix)
				dbIterateOverKeys(prefix, "", func(k string, v []byte) {
					db.Delete([]byte(k), nil)
					publishChange(flow.Tag{k, nil})
				})
			case "<range>":
				prefix := tag.Msg.(string)
				glog.V(3).Infoln("range", prefix)
				w.Out.Send(m)
				dbIterateOverKeys(prefix, "", func(k string, v []byte) {
					var any interface{}
					err := json.Unmarshal(v, &any)
					flow.Check(err)
					w.Out.Send(flow.Tag{k, any})
				})
			case "<register>":
				dbRegister(tag.Msg.(string))
				// publishChange(tag) // TODO: why was this being sent out?
			default:
				if strings.HasPrefix(tag.Tag, "<") {
					w.Out.Send(m) // pass on other tags without processing
				} else {
					dbPut(tag.Tag, tag.Msg)
					publishChange(tag)
				}
			}
		} else {
			w.Out.Send(m)
		}
	}
}

// Start the MQTT server.
func (w *RemoteMQTTServer) Run() {
	if glog.V(2) {
		glog.Infoln("RemoteMQTTBroker Run begins...")
	}

	port := getInputOrConfigwithDefault(w.Port, "MQTT_PORT", ":1883")

	//TODO: Perhaps add a 'real' server check by making an MQTT Client connection.
	conn, err := net.Dial("tcp", port)
	if err != nil { //This gives a more specific error log, scoped to file
		glog.Errorln("Error connecting to MQTT Server:", err)
	}
	flow.Check(err) //And then let flow panic.

	defer conn.Close()

	Done := make(chan bool)
	w.PortOut.Send(port)
	<-Done //we dont really need this!
}

// Start processing text lines to and from the serial interface.
// Send a bool to adjust RTS or an int to pulse DTR for that many milliseconds.
// Registers as "SerialPort".
func (w *SerialPort) Run() {
	if port, ok := <-w.Port; ok {
		opt := rs232.Options{BitRate: 57600, DataBits: 8, StopBits: 1}
		dev, err := rs232.Open(port.(string), opt)
		flow.Check(err)

		// try to avoid kernel panics due to that wretched buggy FTDI driver!
		// defer func() {
		// 	time.Sleep(time.Second)
		// 	dev.Close()
		// }()
		// time.Sleep(time.Second)

		// separate process to copy data out to the serial port
		go func() {
			for m := range w.To {
				switch v := m.(type) {
				case string:
					dev.Write([]byte(v + "\n"))
				case []byte:
					dev.Write(v)
				case int:
					dev.SetDTR(true) // pulse DTR to reset
					time.Sleep(time.Duration(v) * time.Millisecond)
					dev.SetDTR(false)
				case bool:
					dev.SetRTS(v)
				}
			}
		}()

		scanner := bufio.NewScanner(dev)
		for scanner.Scan() {
			w.From.Send(scanner.Text())
		}
	}
}

// Start decoding radioBlip packets.
func (w *RadioBlip) Run() {
	for m := range w.In {
		if v, ok := m.([]byte); ok && len(v) >= 4 {
			buf := bytes.NewBuffer(v[1:])
			var ping uint32
			err := binary.Read(buf, binary.LittleEndian, &ping)
			flow.Check(err)

			result := map[string]int{
				"<reading>": 1,
				"ping":      int(ping),
				"age":       int(ping / (86400 / 64)),
			}

			if len(v) >= 8 {
				result["tag"] = int(v[5] & 0x7F)
				result["vpre"] = 50 + int(v[6])
				if v[5]&0x80 != 0 {
					// if high bit of id is set, this is a boost node
					// reporting its battery -  this is ratiometric
					// (proportional) w.r.t. the "vpre" just measured
					result["vbatt"] = result["vpre"] * int(v[7]) / 255
				} else if v[7] != 0 {
					// in the non-boost case, the second value is vcc
					// after the previous transmit -  this is always set,
					// except in the first transmission after power-up
					result["vpost"] = 50 + int(v[7])
				}
			}

			m = result
		}

		w.Out.Send(m)
	}
}

//Run is the main RadioBlippers gadget entry point.
//This gadget is used to simulate 1 to 30 radioBlip sketches on a specific band/group
//You may incorpoate this Gadget multiple times using different band/group combinations.
//Note: does NOT currently simulate the 'contention' issues that can be experienced on a real RF network.
//Use this to establish numerous 'fake' nodes on a netgroup. Don't forget to add them
//to your node/driver cross reference lookup tables.
func (g *RadioBlippers) Run() {

	band := int(-1)
	group := int(0)

	nodes := make(map[string]*RadioBlipper)

	//read params
	for param := range g.Param {

		p := param.(flow.Tag)

		switch p.Tag {
		case "band":
			band = int(p.Msg.(float64))
		case "group":
			group = int(p.Msg.(float64))
		case "node":
			node := int(p.Msg.(float64))
			if !(node >= 1 && node <= 30) {
				flow.Check(errors.New(fmt.Sprintf("Node %d is out of range 1-30", node)))
				continue
			}
			nodes[fmt.Sprintf("%d", node)] = &RadioBlipper{node, 0}
		}

	}

	if _, ok := radioBands[band]; !ok {
		flow.Check(errors.New(fmt.Sprintf("Band unsupported:%d (433,868,915)", band)))
	}

	if group < 1 || group > 250 {
		flow.Check(errors.New(fmt.Sprintf("Group unsupported:%d (1-250)", group)))
	}

	if len(nodes) == 0 {
		flow.Check(errors.New(fmt.Sprintf("No nodes loaded")))
	}

	<-time.After(time.Millisecond * 500)
	g.From.Send(fmt.Sprintf("[RF12demo.10] _ i31* g%d @ %d MHz", group, band)) //we immitate a collector on node 31

	receiver := time.NewTicker(1 * time.Minute)

	for {
		select {

		case <-receiver.C: //simulate RFDEMO incomming - radioBlips every 1 min
			//we send output messages that simulate the RadioBlip sketch via RF12Demo
			for _, v := range nodes {
				v.Next()
				buf := new(bytes.Buffer)
				_ = binary.Write(buf, binary.LittleEndian, v.payload)

				bytes := buf.Bytes()

				msg := fmt.Sprintf("OK %d %d %d %d %d", v.int, bytes[0], bytes[1], bytes[2], bytes[3])
				<-time.After(time.Millisecond * time.Duration(v.int*500)) //vary how quickly they come in over the minute
				g.From.Send(msg)

			}

		}

	}

}