From 5cfb34a151518f0919a3da000a4c78c0d3dd3564 Mon Sep 17 00:00:00 2001 From: Lewis Marshall Date: Sun, 18 Jun 2017 12:13:49 +0200 Subject: [PATCH 01/10] p2p/simulations: Fix stopping exec nodes Signed-off-by: Lewis Marshall --- p2p/simulations/adapters/exec.go | 15 ++++----------- 1 file changed, 4 insertions(+), 11 deletions(-) diff --git a/p2p/simulations/adapters/exec.go b/p2p/simulations/adapters/exec.go index 319e4353a2..81267e097b 100644 --- a/p2p/simulations/adapters/exec.go +++ b/p2p/simulations/adapters/exec.go @@ -183,25 +183,18 @@ func (n *ExecNode) Start(snapshots map[string][]byte) (err error) { // read the WebSocket address from the stderr logs var wsAddr string - errC := make(chan error) + wsAddrC := make(chan string) go func() { s := bufio.NewScanner(stderrR) for s.Scan() { if strings.Contains(s.Text(), "WebSocket endpoint opened:") { - wsAddr = wsAddrPattern.FindString(s.Text()) - break + wsAddrC <- wsAddrPattern.FindString(s.Text()) } } - select { - case errC <- s.Err(): - default: - } }() select { - case err := <-errC: - if err != nil { - return fmt.Errorf("error reading WebSocket address from stderr: %s", err) - } else if wsAddr == "" { + case wsAddr = <-wsAddrC: + if wsAddr == "" { return errors.New("failed to read WebSocket address from stderr") } case <-time.After(10 * time.Second): From f07952944503a518f15b35c56a136de7475056e8 Mon Sep 17 00:00:00 2001 From: Lewis Marshall Date: Sun, 18 Jun 2017 12:14:34 +0200 Subject: [PATCH 02/10] p2p/simulations: Don't disconnect nodes before stopping Signed-off-by: Lewis Marshall --- p2p/simulations/network.go | 8 -------- 1 file changed, 8 deletions(-) diff --git a/p2p/simulations/network.go b/p2p/simulations/network.go index 5be6f45de3..2e6f24d514 100644 --- a/p2p/simulations/network.go +++ b/p2p/simulations/network.go @@ -589,14 +589,6 @@ func (self *Network) getConn(oneID, otherID discover.NodeID) *Conn { } func (self *Network) Shutdown() { - // disconnect all nodes - for _, conn := range self.Conns { - log.Debug(fmt.Sprintf("disconnecting %s from %s", conn.One.TerminalString(), conn.Other.TerminalString())) - if err := self.Disconnect(conn.One, conn.Other); err != nil { - log.Warn(fmt.Sprintf("error disconnecting %s from %s", conn.One.TerminalString(), conn.Other.TerminalString()), "err", err) - } - } - // stop all nodes for _, node := range self.Nodes { log.Debug(fmt.Sprintf("stopping node %s", node.ID().TerminalString())) From be95e9ecb38da2fea97ed9777dccbc6abfdc01b0 Mon Sep 17 00:00:00 2001 From: Lewis Marshall Date: Sun, 11 Jun 2017 21:14:32 +0100 Subject: [PATCH 03/10] p2p/simulations: Set node ID if not set Signed-off-by: Lewis Marshall --- p2p/simulations/network.go | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/p2p/simulations/network.go b/p2p/simulations/network.go index 2e6f24d514..552e78ebad 100644 --- a/p2p/simulations/network.go +++ b/p2p/simulations/network.go @@ -201,6 +201,11 @@ func (self *Network) NewNode() (*Node, error) { func (self *Network) NewNodeWithConfig(conf *adapters.NodeConfig) (*Node, error) { self.lock.Lock() defer self.lock.Unlock() + if conf.ID == (discover.NodeID{}) { + c := adapters.RandomNodeConfig() + conf.ID = c.ID + conf.PrivateKey = c.PrivateKey + } id := conf.ID if conf.Name == "" { conf.Name = fmt.Sprintf("node%02d", len(self.Nodes)+1) From b56a7fb5ef32aa40a25d1bc39a698e72921f681b Mon Sep 17 00:00:00 2001 From: Lewis Marshall Date: Wed, 14 Jun 2017 11:05:19 +0200 Subject: [PATCH 04/10] p2p/simulations/examples: Log errors Signed-off-by: Lewis Marshall --- p2p/simulations/examples/connectivity.go | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/p2p/simulations/examples/connectivity.go b/p2p/simulations/examples/connectivity.go index 95b7c8be7e..4e74211787 100644 --- a/p2p/simulations/examples/connectivity.go +++ b/p2p/simulations/examples/connectivity.go @@ -67,7 +67,9 @@ func main() { } log.Info("starting simulation server on 0.0.0.0:8888...") - http.ListenAndServe(":8888", simulations.NewServer(config)) + if err := http.ListenAndServe(":8888", simulations.NewServer(config)); err != nil { + log.Crit("error starting simulation server", "err", err) + } } // pingPongService runs a ping-pong protocol between nodes where each node From 8ac9af986c67b0ea4153a42915ffee59984fda3c Mon Sep 17 00:00:00 2001 From: Lewis Marshall Date: Wed, 14 Jun 2017 11:05:45 +0200 Subject: [PATCH 05/10] p2p/simulations: Ensure nodes don't already exist Signed-off-by: Lewis Marshall --- p2p/simulations/network.go | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/p2p/simulations/network.go b/p2p/simulations/network.go index 552e78ebad..9040fcfec4 100644 --- a/p2p/simulations/network.go +++ b/p2p/simulations/network.go @@ -207,9 +207,15 @@ func (self *Network) NewNodeWithConfig(conf *adapters.NodeConfig) (*Node, error) conf.PrivateKey = c.PrivateKey } id := conf.ID + if node := self.getNode(id); node != nil { + return nil, fmt.Errorf("node already exists: %q", id) + } if conf.Name == "" { conf.Name = fmt.Sprintf("node%02d", len(self.Nodes)+1) } + if node := self.getNodeByName(conf.Name); node != nil { + return nil, fmt.Errorf("node already exists: %q", conf.Name) + } if len(conf.Services) == 0 { conf.Services = []string{self.DefaultService} } @@ -533,6 +539,10 @@ func (self *Network) GetNode(id discover.NodeID) *Node { func (self *Network) GetNodeByName(name string) *Node { self.lock.Lock() defer self.lock.Unlock() + return self.getNodeByName(name) +} + +func (self *Network) getNodeByName(name string) *Node { for _, node := range self.Nodes { if node.Config.Name == name { return node From e32197f998dd18aa378ca75df891d3e8fe822925 Mon Sep 17 00:00:00 2001 From: Lewis Marshall Date: Wed, 14 Jun 2017 11:06:32 +0200 Subject: [PATCH 06/10] p2p/simulations: Don't log errors for duplicate events Signed-off-by: Lewis Marshall --- p2p/simulations/network.go | 16 ++++------------ 1 file changed, 4 insertions(+), 12 deletions(-) diff --git a/p2p/simulations/network.go b/p2p/simulations/network.go index 9040fcfec4..df08c41bcf 100644 --- a/p2p/simulations/network.go +++ b/p2p/simulations/network.go @@ -347,21 +347,13 @@ func (self *Network) watchPeerEvents(id discover.NodeID, events chan *p2p.PeerEv peer := event.Peer switch event.Type { case p2p.PeerEventTypeAdd: - if err := self.DidConnect(id, peer); err != nil { - log.Error(fmt.Sprintf("error generating connection up event %s => %s", id.TerminalString(), peer.TerminalString()), "err", err) - } + self.DidConnect(id, peer) case p2p.PeerEventTypeDrop: - if err := self.DidDisconnect(id, peer); err != nil { - log.Error(fmt.Sprintf("error generating connection down event %s => %s", id.TerminalString(), peer.TerminalString()), "err", err) - } + self.DidDisconnect(id, peer) case p2p.PeerEventTypeMsgSend: - if err := self.DidSend(id, peer, *event.MsgCode); err != nil { - log.Error(fmt.Sprintf("error generating msg send event %s => %s", id.TerminalString(), peer.TerminalString()), "err", err) - } + self.DidSend(id, peer, *event.MsgCode) case p2p.PeerEventTypeMsgRecv: - if err := self.DidReceive(peer, id, *event.MsgCode); err != nil { - log.Error(fmt.Sprintf("error generating msg receive event %s => %s", peer.TerminalString(), id.TerminalString()), "err", err) - } + self.DidReceive(peer, id, *event.MsgCode) } case err := <-sub.Err(): if err != nil { From 8e17ee44f0a3246b633134107039b8467e131883 Mon Sep 17 00:00:00 2001 From: Lewis Marshall Date: Sun, 11 Jun 2017 21:13:58 +0100 Subject: [PATCH 07/10] p2psim: Use P2PSIM_NETWORK environment variable Signed-off-by: Lewis Marshall --- p2p/simulations/cmd/p2psim/main.go | 92 ++++++++++++++++-------------- p2p/simulations/examples/p2psim.sh | 7 ++- 2 files changed, 54 insertions(+), 45 deletions(-) diff --git a/p2p/simulations/cmd/p2psim/main.go b/p2p/simulations/cmd/p2psim/main.go index 4c7f427fa3..bf00aa768c 100644 --- a/p2p/simulations/cmd/p2psim/main.go +++ b/p2p/simulations/cmd/p2psim/main.go @@ -26,6 +26,7 @@ package main import ( "context" "encoding/json" + "errors" "fmt" "io" "os" @@ -39,7 +40,10 @@ import ( "gopkg.in/urfave/cli.v1" ) -var client *simulations.Client +var ( + client *simulations.Client + networkID string +) func main() { app := cli.NewApp() @@ -109,18 +113,32 @@ func main() { Name: "node", Usage: "manage simulation nodes", Action: listNodes, + Flags: []cli.Flag{ + cli.StringFlag{ + Name: "network", + Usage: "simulation network", + }, + }, + Before: func(ctx *cli.Context) error { + networkID = ctx.GlobalString("network") + if networkID == "" { + networkID = os.Getenv("P2PSIM_NETWORK") + } + if networkID == "" { + return errors.New("missing network, set with --network or P2PSIM_NETWORK") + } + return nil + }, Subcommands: []cli.Command{ { - Name: "list", - ArgsUsage: "", - Usage: "list nodes", - Action: listNodes, + Name: "list", + Usage: "list nodes", + Action: listNodes, }, { - Name: "create", - ArgsUsage: "", - Usage: "create a node", - Action: createNode, + Name: "create", + Usage: "create a node", + Action: createNode, Flags: []cli.Flag{ cli.StringFlag{ Name: "config", @@ -131,37 +149,37 @@ func main() { }, { Name: "show", - ArgsUsage: " ", + ArgsUsage: "", Usage: "show node information", Action: showNode, }, { Name: "start", - ArgsUsage: " ", + ArgsUsage: "", Usage: "start a node", Action: startNode, }, { Name: "stop", - ArgsUsage: " ", + ArgsUsage: "", Usage: "stop a node", Action: stopNode, }, { Name: "connect", - ArgsUsage: " ", + ArgsUsage: " ", Usage: "connect a node to a peer node", Action: connectNode, }, { Name: "disconnect", - ArgsUsage: " ", + ArgsUsage: " ", Usage: "disconnect a node from a peer node", Action: disconnectNode, }, { Name: "rpc", - ArgsUsage: " []", + ArgsUsage: " []", Usage: "call a node RPC method", Action: rpcNode, Flags: []cli.Flag{ @@ -280,11 +298,9 @@ func loadSnapshot(ctx *cli.Context) error { } func listNodes(ctx *cli.Context) error { - args := ctx.Args() - if len(args) != 1 { + if len(ctx.Args()) != 0 { return cli.ShowCommandHelp(ctx, ctx.Command.Name) } - networkID := args[0] nodes, err := client.GetNodes(networkID) if err != nil { return err @@ -307,11 +323,9 @@ func protocolList(node *p2p.NodeInfo) []string { } func createNode(ctx *cli.Context) error { - args := ctx.Args() - if len(args) != 1 { + if len(ctx.Args()) != 0 { return cli.ShowCommandHelp(ctx, ctx.Command.Name) } - networkID := args[0] config := &adapters.NodeConfig{} if err := json.Unmarshal([]byte(ctx.String("config")), config); err != nil { return err @@ -326,11 +340,10 @@ func createNode(ctx *cli.Context) error { func showNode(ctx *cli.Context) error { args := ctx.Args() - if len(args) != 2 { + if len(args) != 1 { return cli.ShowCommandHelp(ctx, ctx.Command.Name) } - networkID := args[0] - nodeName := args[1] + nodeName := args[0] node, err := client.GetNode(networkID, nodeName) if err != nil { return err @@ -352,11 +365,10 @@ func showNode(ctx *cli.Context) error { func startNode(ctx *cli.Context) error { args := ctx.Args() - if len(args) != 2 { + if len(args) != 1 { return cli.ShowCommandHelp(ctx, ctx.Command.Name) } - networkID := args[0] - nodeName := args[1] + nodeName := args[0] if err := client.StartNode(networkID, nodeName); err != nil { return err } @@ -366,11 +378,10 @@ func startNode(ctx *cli.Context) error { func stopNode(ctx *cli.Context) error { args := ctx.Args() - if len(args) != 2 { + if len(args) != 1 { return cli.ShowCommandHelp(ctx, ctx.Command.Name) } - networkID := args[0] - nodeName := args[1] + nodeName := args[0] if err := client.StopNode(networkID, nodeName); err != nil { return err } @@ -380,12 +391,11 @@ func stopNode(ctx *cli.Context) error { func connectNode(ctx *cli.Context) error { args := ctx.Args() - if len(args) != 3 { + if len(args) != 2 { return cli.ShowCommandHelp(ctx, ctx.Command.Name) } - networkID := args[0] - nodeName := args[1] - peerName := args[2] + nodeName := args[0] + peerName := args[1] if err := client.ConnectNode(networkID, nodeName, peerName); err != nil { return err } @@ -395,12 +405,11 @@ func connectNode(ctx *cli.Context) error { func disconnectNode(ctx *cli.Context) error { args := ctx.Args() - if len(args) != 3 { + if len(args) != 2 { return cli.ShowCommandHelp(ctx, ctx.Command.Name) } - networkID := args[0] - nodeName := args[1] - peerName := args[2] + nodeName := args[0] + peerName := args[1] if err := client.DisconnectNode(networkID, nodeName, peerName); err != nil { return err } @@ -410,12 +419,11 @@ func disconnectNode(ctx *cli.Context) error { func rpcNode(ctx *cli.Context) error { args := ctx.Args() - if len(args) < 3 { + if len(args) < 2 { return cli.ShowCommandHelp(ctx, ctx.Command.Name) } - networkID := args[0] - nodeName := args[1] - method := args[2] + nodeName := args[0] + method := args[1] rpcClient, err := client.RPCClient(context.Background(), networkID, nodeName) if err != nil { return err diff --git a/p2p/simulations/examples/p2psim.sh b/p2p/simulations/examples/p2psim.sh index 7c83285cc7..b67a91ce0c 100755 --- a/p2p/simulations/examples/p2psim.sh +++ b/p2p/simulations/examples/p2psim.sh @@ -13,14 +13,15 @@ main() { p2psim network create --config '{"id": "example", "default_service": "ping-pong"}' info "creating 10 nodes" + export P2PSIM_NETWORK="example" for i in $(seq 1 10); do - p2psim node create "example" - p2psim node start "example" "$(node_name $i)" + p2psim node create + p2psim node start "$(node_name $i)" done info "connecting node01 to all other nodes" for i in $(seq 2 10); do - p2psim node connect "example" "node01" "$(node_name $i)" + p2psim node connect "node01" "$(node_name $i)" done info "done" From 9627a75ce0adb768493d52ba16ff930ddc9cbc8a Mon Sep 17 00:00:00 2001 From: Lewis Marshall Date: Mon, 12 Jun 2017 19:27:28 +0200 Subject: [PATCH 08/10] p2psim: Use explicit flags rather than JSON config Signed-off-by: Lewis Marshall --- p2p/simulations/cmd/p2psim/main.go | 51 +++++++++++++++++++++++------- p2p/simulations/examples/p2psim.sh | 6 ++-- 2 files changed, 42 insertions(+), 15 deletions(-) diff --git a/p2p/simulations/cmd/p2psim/main.go b/p2p/simulations/cmd/p2psim/main.go index bf00aa768c..012611c5ad 100644 --- a/p2p/simulations/cmd/p2psim/main.go +++ b/p2p/simulations/cmd/p2psim/main.go @@ -33,7 +33,9 @@ import ( "strings" "text/tabwriter" + "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/p2p" + "github.com/ethereum/go-ethereum/p2p/discover" "github.com/ethereum/go-ethereum/p2p/simulations" "github.com/ethereum/go-ethereum/p2p/simulations/adapters" "github.com/ethereum/go-ethereum/rpc" @@ -77,9 +79,14 @@ func main() { Action: createNetwork, Flags: []cli.Flag{ cli.StringFlag{ - Name: "config", - Value: "{}", - Usage: "JSON encoded network config", + Name: "id", + Value: "", + Usage: "network ID", + }, + cli.StringFlag{ + Name: "default-service", + Value: "", + Usage: "default service", }, }, }, @@ -141,9 +148,19 @@ func main() { Action: createNode, Flags: []cli.Flag{ cli.StringFlag{ - Name: "config", - Value: "{}", - Usage: "JSON encoded node config", + Name: "name", + Value: "", + Usage: "node name", + }, + cli.StringFlag{ + Name: "services", + Value: "", + Usage: "node services (comma separated)", + }, + cli.StringFlag{ + Name: "key", + Value: "", + Usage: "node private key (hex encoded)", }, }, }, @@ -216,9 +233,9 @@ func createNetwork(ctx *cli.Context) error { if len(ctx.Args()) != 0 { return cli.ShowCommandHelp(ctx, ctx.Command.Name) } - config := &simulations.NetworkConfig{} - if err := json.Unmarshal([]byte(ctx.String("config")), config); err != nil { - return err + config := &simulations.NetworkConfig{ + ID: ctx.String("id"), + DefaultService: ctx.String("default-service"), } network, err := client.CreateNetwork(config) if err != nil { @@ -326,9 +343,19 @@ func createNode(ctx *cli.Context) error { if len(ctx.Args()) != 0 { return cli.ShowCommandHelp(ctx, ctx.Command.Name) } - config := &adapters.NodeConfig{} - if err := json.Unmarshal([]byte(ctx.String("config")), config); err != nil { - return err + config := &adapters.NodeConfig{ + Name: ctx.String("name"), + } + if key := ctx.String("key"); key != "" { + privKey, err := crypto.HexToECDSA(key) + if err != nil { + return err + } + config.ID = discover.PubkeyID(&privKey.PublicKey) + config.PrivateKey = privKey + } + if services := ctx.String("services"); services != "" { + config.Services = strings.Split(services, ",") } node, err := client.CreateNode(networkID, config) if err != nil { diff --git a/p2p/simulations/examples/p2psim.sh b/p2p/simulations/examples/p2psim.sh index b67a91ce0c..468d66d4e3 100755 --- a/p2p/simulations/examples/p2psim.sh +++ b/p2p/simulations/examples/p2psim.sh @@ -10,12 +10,12 @@ main() { fi info "creating the example network" - p2psim network create --config '{"id": "example", "default_service": "ping-pong"}' + export P2PSIM_NETWORK="example" + p2psim network create --id "${P2PSIM_NETWORK}" info "creating 10 nodes" - export P2PSIM_NETWORK="example" for i in $(seq 1 10); do - p2psim node create + p2psim node create --name "$(node_name $i)" --services "ping-pong" p2psim node start "$(node_name $i)" done From 898f0c1b154eb496d19f21632b95226ea7eb57c1 Mon Sep 17 00:00:00 2001 From: Lewis Marshall Date: Sun, 18 Jun 2017 16:13:05 +0200 Subject: [PATCH 09/10] p2p/simulations: Fix the DockerAdapter Signed-off-by: Lewis Marshall --- p2p/simulations/adapters/docker.go | 20 ++++++++++++++++---- p2p/simulations/adapters/exec.go | 13 ++++++++++--- 2 files changed, 26 insertions(+), 7 deletions(-) diff --git a/p2p/simulations/adapters/docker.go b/p2p/simulations/adapters/docker.go index eb20ac3318..894057256e 100644 --- a/p2p/simulations/adapters/docker.go +++ b/p2p/simulations/adapters/docker.go @@ -13,6 +13,7 @@ import ( "github.com/docker/docker/pkg/reexec" "github.com/ethereum/go-ethereum/node" + "github.com/ethereum/go-ethereum/p2p/discover" ) // DockerAdapter is a NodeAdapter which runs nodes inside Docker containers. @@ -20,7 +21,9 @@ import ( // A Docker image is built which contains the current binary at /bin/p2p-node // which when executed runs the underlying service (see the description // of the execP2PNode function for more details) -type DockerAdapter struct{} +type DockerAdapter struct { + ExecAdapter +} // NewDockerAdapter builds the p2p-node Docker image containing the current // binary and returns a DockerAdapter @@ -33,7 +36,11 @@ func NewDockerAdapter() (*DockerAdapter, error) { return nil, err } - return &DockerAdapter{}, nil + return &DockerAdapter{ + ExecAdapter{ + nodes: make(map[discover.NodeID]*ExecNode), + }, + }, nil } // Name returns the name of the adapter for logging purpoeses @@ -58,17 +65,22 @@ func (d *DockerAdapter) NewNode(config *NodeConfig) (Node, error) { Node: config, } conf.Stack.DataDir = "/data" + conf.Stack.WSHost = "0.0.0.0" + conf.Stack.WSOrigins = []string{"*"} + conf.Stack.WSExposeAll = true conf.Stack.P2P.EnableMsgEvents = true conf.Stack.P2P.NoDiscovery = true conf.Stack.P2P.NAT = nil node := &DockerNode{ ExecNode: ExecNode{ - ID: config.ID, - Config: conf, + ID: config.ID, + Config: conf, + adapter: &d.ExecAdapter, }, } node.newCmd = node.dockerCommand + d.ExecAdapter.nodes[node.ID] = &node.ExecNode return node, nil } diff --git a/p2p/simulations/adapters/exec.go b/p2p/simulations/adapters/exec.go index 81267e097b..f7f047ef51 100644 --- a/p2p/simulations/adapters/exec.go +++ b/p2p/simulations/adapters/exec.go @@ -347,17 +347,24 @@ func execP2PNode() { conf.Stack.P2P.PrivateKey = conf.Node.PrivateKey // use explicit IP address in ListenAddr so that Enode URL is usable - if strings.HasPrefix(conf.Stack.P2P.ListenAddr, ":") { + externalIP := func() string { addrs, err := net.InterfaceAddrs() if err != nil { log.Crit("error getting IP address", "err", err) } for _, addr := range addrs { if ip, ok := addr.(*net.IPNet); ok && !ip.IP.IsLoopback() { - conf.Stack.P2P.ListenAddr = ip.IP.String() + conf.Stack.P2P.ListenAddr - break + return ip.IP.String() } } + log.Crit("unable to determine explicit IP address") + return "" + } + if strings.HasPrefix(conf.Stack.P2P.ListenAddr, ":") { + conf.Stack.P2P.ListenAddr = externalIP() + conf.Stack.P2P.ListenAddr + } + if conf.Stack.WSHost == "0.0.0.0" { + conf.Stack.WSHost = externalIP() } // initialize the devp2p stack From ca8421c336d92e2a3991fb036d0ba323b95c1fa2 Mon Sep 17 00:00:00 2001 From: Lewis Marshall Date: Sun, 18 Jun 2017 16:26:36 +0200 Subject: [PATCH 10/10] p2p/simulations: Add live node stop events Signed-off-by: Lewis Marshall --- p2p/simulations/network.go | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/p2p/simulations/network.go b/p2p/simulations/network.go index df08c41bcf..706c77dab2 100644 --- a/p2p/simulations/network.go +++ b/p2p/simulations/network.go @@ -337,7 +337,16 @@ func (self *Network) startWithSnapshots(id discover.NodeID, snapshots map[string } func (self *Network) watchPeerEvents(id discover.NodeID, events chan *p2p.PeerEvent, sub event.Subscription) { - defer sub.Unsubscribe() + defer func() { + sub.Unsubscribe() + + // assume the node is now down + self.lock.Lock() + node := self.getNode(id) + node.Up = false + self.lock.Unlock() + self.events.Send(NewEvent(node)) + }() for { select { case event, ok := <-events: