func ExampleNestedCircuit() {
	g1 := flow.NewCircuit()
	g1.Add("p", "Pipe")
	g1.Label("In", "p.In")
	g1.Label("Out", "p.Out")

	g2 := flow.NewCircuit()
	g2.Add("p", "Pipe")
	g2.Label("In", "p.In")
	g2.Label("Out", "p.Out")

	g3 := flow.NewCircuit()
	g3.AddCircuitry("g1", g1)
	g3.AddCircuitry("g2", g2)
	g3.Connect("g1.Out", "g2.In", 0)
	g3.Label("In", "g1.In")
	g3.Label("Out", "g2.Out")

	g := flow.NewCircuit()
	g.Add("p1", "Pipe")
	g.AddCircuitry("g", g3)
	g.Add("p2", "Pipe")
	g.Connect("p1.Out", "g.In", 0)
	g.Connect("g.Out", "p2.In", 0)
	g.Feed("p1.In", "abc")
	g.Feed("p1.In", "def")
	g.Feed("p1.In", "ghi")
	g.Run()
	// Output:
	// Lost string: abc
	// Lost string: def
	// Lost string: ghi
}

func init() {
	// Waiter can be defined in terms of other gadgets, but where to put this?
	flow.Registry["Waiter"] = func() flow.Circuitry {
		c := flow.NewCircuit()
		c.Add("s", "Sink")
		c.Add("c", "Concat3")
		c.Connect("s.Out", "c.In1", 0)
		c.Label("Gate", "s.In")
		c.Label("In", "c.In2")
		c.Label("Out", "c.Out")
		return c
	}
	// Attach returns items from a database range and follows up with updates.
	flow.Registry["Attach"] = func() flow.Circuitry {
		c := flow.NewCircuit()
		c.Add("fan", "FanOut")
		c.Add("sub", "DataSub")
		c.Add("tag", "AddTag")
		c.Add("db", "LevelDB")
		c.Add("cat", "Concat3")
		c.Connect("fan.Out:sub", "sub.In", 0)
		c.Connect("fan.Out:tag", "tag.In", 0)
		c.Connect("tag.Out", "db.In", 0)
		c.Connect("db.Out", "cat.In1", 0)
		c.Connect("sub.Out", "cat.In3", 100) // with buffering
		c.Feed("tag.Tag", "<range>")
		c.Feed("cat.In2", flow.Tag{"<sync>", "attach"})
		c.Label("In", "fan.In")
		c.Label("Out", "cat.Out")
		return c
	}

	flow.Registry["help"] = func() flow.Circuitry { return new(helpCmd) }
	Help["help"] = `Show this help text with a list of commands.`

	flow.Registry["info"] = func() flow.Circuitry { return new(infoCmd) }
	Help["info"] = `Show the list of registered gadgets and circuits.`

	// database
	Help["dbdump"] = `Dump (part of) the contents of the database to stdout.`
	Help["dbexport"] = `Export (part of) the database as JSON to stdout.`
	Help["dbimport"] = `Import a JSON file, as generated by "export".`
	Help["dbget"] = `Get one entry from the database.`
	Help["dbput"] = `Store or delete an entry in the database.`
	Help["dbkeys"] = `List the set of (sub-) keys in the database.`
	// network
	Help["mqttpub"] = `Publish a message on a topic using MQTT.`
	Help["mqttsub"] = `Subscribe to one or more topics using MQTT.`
}

func (g *RpcHandler) handleRpcRequest(cmd string, seq int, args []interface{}) (reply []interface{}) {
	glog.Infoln("rpc", cmd, args)
	if cmd == "echo" {
		return []interface{}{seq, "", args}
	}

	defer func() {
		errMsg := ""
		switch v := recover().(type) {
		case nil:
			// no error
		case string:
			errMsg = v
		case error:
			errMsg = v.Error()
		default:
			errMsg = fmt.Sprintf("%T: %v", v, v)
		}
		if errMsg != "" {
			glog.Warningln("rpc-error", cmd, args, errMsg)
			reply = []interface{}{seq, errMsg}
		}
	}()

	// FIXME: hack alert, this is special-cased to deal with the database
	switch cmd {
	case "db-get":
		result := database.Get(args[0].(string))
		return []interface{}{seq, "", result}
	case "db-keys":
		result := database.Keys(args[0].(string))
		return []interface{}{seq, "", result}
	case "db-put":
		var value interface{}
		if len(args) > 1 {
			value = args[1]
		}
		database.Put(args[0].(string), value)
		return nil
	}

	// if there's registered circuit for cmd, set it up and return as a stream
	if _, ok := flow.Registry[cmd]; ok && len(args) == 1 {
		c := flow.NewCircuit()
		c.Add("x", cmd)
		c.AddCircuitry("y", &streamRpcResults{seqNum: seq, replies: g})
		c.Connect("x.Out", "y.In", 0)
		for k, v := range args[0].(map[string]interface{}) {
			c.Feed("x."+k, tryToConvertToTag(v))
		}
		go func() {
			defer flow.DontPanic()
			c.Run()
			g.Out.Send([]interface{}{seq, false}) // end streaming
		}()
		return []interface{}{seq, true} // start streaming
	}

	panic(cmd + "?")
}

func ExampleLevelDB() {
	g := flow.NewCircuit()
	g.Add("db", "LevelDB")
	g.Feed("db.In", flow.Tag{"a/b", "123"})
	g.Feed("db.In", flow.Tag{"a/c", "456"})
	g.Feed("db.In", flow.Tag{"<get>", "a/b"})
	g.Feed("db.In", flow.Tag{"<range>", "a/"})
	g.Feed("db.In", flow.Tag{"<keys>", "a/"})
	g.Feed("db.In", flow.Tag{"a/b", nil})
	g.Feed("db.In", flow.Tag{"<get>", "a/b"})
	g.Feed("db.In", flow.Tag{"<keys>", "a/"})
	g.Feed("db.In", flow.Tag{"a/c", nil})
	g.Run()
	// Output:
	// Lost flow.Tag: {<get> a/b}
	// Lost string: 123
	// Lost flow.Tag: {<range> a/}
	// Lost flow.Tag: {a/b 123}
	// Lost flow.Tag: {a/c 456}
	// Lost flow.Tag: {<keys> a/}
	// Lost string: b
	// Lost string: c
	// Lost flow.Tag: {<get> a/b}
	// Lost <nil>: <nil>
	// Lost flow.Tag: {<keys> a/}
	// Lost string: c
}

func ExampleTimeStamp() {
	g := flow.NewCircuit()
	g.Add("t", "TimeStamp")
	g.Run()
	g.Feed("t.In", "abc")
	// produces two lines, the timestamp followed by the "abc" string
}

//wsHandler now used ws.Config as protocol handshake now supported
func wsHandler(ws *websocket.Conn) {
	defer flow.DontPanic()
	defer ws.Close()

	hdr := ws.Request().Header

	// keep track of connected clients for reload broadcasting
	id := hdr.Get("Sec-Websocket-Key")
	wsClients[id] = ws
	defer delete(wsClients, id)

	// the protocol name is used as tag to locate the proper circuit
	//lightbulb: We use the protocol provided by ws, rather than header, as that contains server accepted value
	tag := ws.Config().Protocol[0]

	fmt.Println("WS Protocol Selected:", tag)

	if tag == "" { //no specific protocol, lets opt for 'default' which just echoes (or return with no circuit!)
		tag = "default"
	}

	g := flow.NewCircuit()
	g.AddCircuitry("head", &wsHead{ws: ws})
	g.Add("ws", "WebSocket-"+tag) //the client has negotiated this support
	g.AddCircuitry("tail", &wsTail{ws: ws})
	g.Connect("head.Out", "ws.In", 0)
	g.Connect("ws.Out", "tail.In", 0)
	g.Run()
}

func ExampleClock() {
	// The following example never ends.
	g := flow.NewCircuit()
	g.Add("c", "Clock")
	g.Feed("c.In", "1s")
	g.Run()
}

func ExampleMQTTSub() {
	// The following example requires an MQTT server running on std port 1883.
	g := flow.NewCircuit()
	g.Add("s", "MQTTSub")
	g.Feed("s.Port", ":1883")
	g.Feed("s.In", "#")
	g.Run()
}

func ExampleMQTTPub() {
	// The following example requires an MQTT server running on std port 1883.
	g := flow.NewCircuit()
	g.Add("p", "MQTTPub")
	g.Feed("p.Port", ":1883")
	g.Feed("p.In", flow.Tag{"Hello", "world"})
	g.Run()
}

func ExamplePrinter() {
	g := flow.NewCircuit()
	g.Add("p", "Printer")
	g.Feed("p.In", "hello")
	g.Run()
	// Output:
	// hello
}

func ExamplePipe() {
	g := flow.NewCircuit()
	g.Add("p", "Pipe")
	g.Feed("p.In", 123)
	g.Run()
	// Output:
	// Lost int: 123
}

func ExampleCounter() {
	g := flow.NewCircuit()
	g.Add("c", "Counter")
	g.Feed("c.In", nil)
	g.Run()
	// Output:
	// Lost int: 1
}

func (g *pubCmd) Run() {
	c := flow.NewCircuit()
	c.Add("c", "CmdLine")
	c.Add("p", "MQTTPub")
	c.Connect("a.Out", "p.In", 0)
	c.Feed("c.Type", "skip,tags,json")
	c.Feed("p.Port", ":1883")
	c.Run()
}

func ExampleReadFileJSON() {
	g := flow.NewCircuit()
	g.Add("r", "ReadFileJSON")
	g.Feed("r.In", "example.json")
	g.Run()
	// Output:
	// Lost flow.Tag: {<file> example.json}
	// Lost map[string]interface {}: map[a:123 b:[3 4 5] c:true]
}

func ExampleBinaryFill() {
	g := flow.NewCircuit()
	g.Add("f", "BinaryFill")
	g.Feed("f.In", []byte("abcdef"))
	g.Feed("f.Len", 5)
	g.Run()
	// Output:
	// Lost []uint8: [97 98 99 100 101 102 255 255 255 255]
}

func ExampleCalcCrc16() {
	g := flow.NewCircuit()
	g.Add("c", "CalcCrc16")
	g.Feed("c.In", []byte("abc"))
	g.Run()
	// Output:
	// Lost []uint8: [97 98 99]
	// Lost flow.Tag: {<crc16> 22345}
}

func BenchmarkRepeat0(b *testing.B) {
	g := flow.NewCircuit()
	g.Add("r", "Repeater")
	g.Add("s", "Sink")
	g.Connect("r.Out", "s.In", 0)
	g.Feed("r.In", nil)
	g.Feed("r.Num", b.N)
	g.Run()
}

func ExampleHomePower() {
	g := flow.NewCircuit()
	g.Add("d", "Node-homePower")
	g.Feed("d.In", []byte{9, 213, 11, 68, 235, 151, 90, 99, 6, 88, 198, 136, 89})
	g.Feed("d.In", []byte{9, 213, 11, 68, 235, 153, 90, 84, 6, 88, 198, 136, 89})
	g.Run()
	// Output:
	// Lost map[string]int: map[<reading>:1 c1:3029 p1:78 c2:23191 p2:11009 c3:50776 p3:785]
	// Lost map[string]int: map[<reading>:1 c2:23193 p2:11111]
}

func ExampleTGZByYearGZ() {
	g := flow.NewCircuit()
	g.Add("f", "LogArchiverTGZ")
	g.Feed("f.Param", flow.Tag{"-m", "2006.gz"})
	g.Feed("f.Param", flow.Tag{"-v", true})
	g.Feed("f.In", "log/2014/2014.tar")
	g.Run()
	// Output:
	// Lost string: log/2014/2014.tar.gz

}

func main() {
	flag.Parse() // required, to set up the proper glog configuration
	flow.LoadConfig(defaults, *config)
	flow.DontPanic()

	// register more definitions from a JSON-formatted setup file, if specified
	if s := flow.Config["SETUP_FILE"]; s != "" {
		if err := flow.AddToRegistry(s); err != nil {
			panic(err)
		}
	}

	// if a registered circuit name is given on the command line, run it
	if flag.NArg() > 0 {
		if factory, ok := flow.Registry[flag.Arg(0)]; ok {
			factory().Run()
			return
		}
		fmt.Fprintln(os.Stderr, "Unknown command:", flag.Arg(0))
		os.Exit(1)
	}

	fmt.Printf("Starting webserver for http://%s/\n", flow.Config["HTTP_PORT"])

	// normal startup: save config info in database and start the webserver
	c := flow.NewCircuit()

	// database setup, save current config settings, register init gadget
	c.Add("db", "LevelDB")
	c.Feed("db.In", flow.Tag{"<clear>", "/config/"})
	c.Feed("db.In", flow.Tag{"/config/appName", "JeeBus"})
	c.Feed("db.In", flow.Tag{"/config/configFile", *config})
	for k, v := range flow.Config {
		c.Feed("db.In", flow.Tag{"/config/" + k, v})
	}
	c.Feed("db.In", flow.Tag{"<register>", "/gadget/init"})

	// wait for db to finish, then dispatch to the "init" gadget, if found
	c.Add("wait", "Waiter")
	c.Add("disp", "Dispatcher")
	c.Connect("db.Out", "wait.Gate", 0)
	c.Connect("wait.Out", "disp.In", 0)
	c.Feed("wait.In", flow.Tag{"<dispatch>", "init"})

	// webserver setup
	c.Add("http", "HTTPServer")
	c.Feed("http.Handlers", flow.Tag{"/", flow.Config["APP_DIR"]})
	c.Feed("http.Handlers", flow.Tag{"/base/", flow.Config["BASE_DIR"]})
	c.Feed("http.Handlers", flow.Tag{"/ws", "<websocket>"})

	// start the ball rolling, keep running forever
	c.Add("forever", "Forever")
	c.Run()
}