p2p/simulations: refactor journal and cytoscape bridge

p2p/simulations/cytoscape.go

@@ -0,0 +1,100 @@
+package simulations
+
+import (
+	"bytes"
+	"fmt"
+
+	"github.com/ethereum/go-ethereum/event"
+	"github.com/ethereum/go-ethereum/logger/glog"
+)
+
+type CyData struct {
+	Id     string `json:"id"`
+	Source string `json:"source"`
+	Target string `json:"target"`
+	On     bool   `json:"on"`
+}
+
+type CyElement struct {
+	Data    *CyData `json:"data"`
+	Classes string  `json:"classes"`
+	Group   string  `json:"group"`
+	// selected: false, // whether the element is selected (default false)
+	// selectable: true, // whether the selection state is mutable (default true)
+	// locked: false, // when locked a node's position is immutable (default false)
+	// grabbable: true, // whether the node can be grabbed and moved by the user
+}
+
+// type ConnEntry struct {
+// }
+
+// type MsgEntry struct {
+// }
+
+type Entry struct {
+	Action string      `json:"action"`
+	Type   string      `json:"type"`
+	Object interface{} `json:"object"`
+}
+
+func (self *Entry) Stirng() string {
+	return fmt.Sprintf("<Action: %v, Type: %v, Data: %v>\n", self.Action, self.Type, self.Object)
+}
+
+type CyUpdate struct {
+	Add    []*CyElement `json:"add"`
+	Remove []string     `json:"remove"`
+}
+
+func UpdateCy(j *Journal) *CyUpdate {
+	added := []*CyElement{}
+	removed := []string{}
+	var el *CyElement
+	update := func(ev *event.Event) bool {
+		entry := ev.Data.(*Entry)
+		glog.V(6).Infof("journal entry of %v: %v", ev.Time, entry)
+		switch entry.Type {
+		case "Node":
+			el = &CyElement{Group: "nodes", Data: &CyData{Id: entry.Object.(*SimNode).ID.String()[0:lablen]}}
+		case "Conn":
+			// mutually exclusive directed edge (caller -> callee)
+			source := entry.Object.(*SimConn).Caller.String()[0:lablen]
+			target := entry.Object.(*SimConn).Callee.String()[0:lablen]
+			first := source
+			second := target
+			if bytes.Compare([]byte(first), []byte(second)) > 1 {
+				first = target
+				second = source
+			}
+			id := fmt.Sprintf("%v-%v", first, second)
+			el = &CyElement{Group: "edges", Data: &CyData{Id: id, Source: source, Target: target}}
+		case "Know":
+			// independent directed edge (peer0 registers peer1)
+			source := entry.Object.(*Know).Subject.String()[0:lablen]
+			target := entry.Object.(*Know).Object.String()[0:lablen]
+			id := fmt.Sprintf("%v-%v-%v", source, target, "know")
+			el = &CyElement{Group: "edges", Data: &CyData{Id: id, Source: source, Target: target}}
+		}
+		switch entry.Action {
+		case "Add":
+			added = append(added, el)
+		case "Remove":
+			removed = append(removed, el.Data.Id)
+		case "On":
+			el.Data.On = true
+			added = append(added, el)
+		case "Off":
+			el.Data.On = false
+			removed = append(removed, el.Data.Id)
+		}
+		return true
+	}
+	glog.V(6).Infof("journal read")
+	j.Read(update)
+	glog.V(6).Infof("journal read done")
+
+	return &CyUpdate{
+		Add:    added,
+		Remove: removed,
+	}
+}

p2p/simulations/journal.go

@@ -0,0 +1,272 @@
+package simulations
+
+import (
+	"bytes"
+	"math/rand"
+	"sync"
+	"time"
+
+	"github.com/ethereum/go-ethereum/event"
+	"github.com/ethereum/go-ethereum/logger/glog"
+)
+
+// Journal is an instance of a guaranteed no-loss subscription using event.TypeMux
+// Network components POST events to the TypeMux, which then is read by the journal
+// Each journal belongs to a subscription
+type Journal struct {
+	lock    sync.Mutex
+	counter int
+	cursor  int
+	sub     event.Subscription
+	events  []*event.Event
+}
+
+// func (self *Journal) SnapshotAt(pos int) {}
+
+// NewJournal constructor takes eventer and types to subscribe to
+func NewJournal(eventer *event.TypeMux, types ...interface{}) *Journal {
+	self := &Journal{}
+	self.sub = eventer.Subscribe(types...)
+	go func() {
+		self.Write()
+	}()
+	return self
+}
+
+func (self *Journal) Close() {
+	self.sub.Unsubscribe()
+}
+
+// Write is a forever loop
+func (self *Journal) Write() {
+	for {
+		select {
+		case ev, ok := <-self.sub.Chan():
+			if !ok {
+				return
+			}
+			self.append(ev)
+		}
+	}
+}
+
+func (self *Journal) append(evs ...*event.Event) {
+	self.lock.Lock()
+	defer self.lock.Unlock()
+	self.events = append(self.events, evs...)
+}
+
+func (self *Journal) WaitEntries(n int) {
+	for self.NewEntries() < n {
+	}
+}
+
+func (self *Journal) NewEntries() int {
+	self.lock.Lock()
+	defer self.lock.Unlock()
+	return len(self.events) - self.cursor
+}
+
+func (self *Journal) Read(f func(*event.Event) bool) (read int, err error) {
+	self.lock.Lock()
+	defer self.lock.Unlock()
+	for self.cursor < len(self.events) && f(self.events[self.cursor]) {
+		read++
+		self.cursor++
+	}
+	self.reset(self.cursor)
+	return read, nil
+}
+
+func (self *Journal) Reset(n int) {
+	self.lock.Lock()
+	defer self.lock.Unlock()
+	self.reset(n)
+}
+
+func (self *Journal) reset(n int) {
+	if n > self.counter {
+		n = self.counter
+	}
+	self.events = self.events[self.cursor:]
+	self.cursor = 0
+}
+
+func (self *Journal) Counter() int {
+	self.lock.Lock()
+	defer self.lock.Unlock()
+	return self.counter
+}
+
+// type History()
+
+func (self *Journal) Cursor() int {
+	self.lock.Lock()
+	defer self.lock.Unlock()
+	return self.cursor
+}
+
+// deltas: changes in the number of cumulative actions: non-negative integers.
+// base unit is the fixed minimal interval  between two measurements (time quantum)
+// acceleration : to slow down you just set the base unit higher.
+// to speed up: skip x number of base units
+// frequency: given as the (constant or average) number of base units between measurements
+// if resolution is expressed as the inverse of frequency  = preserved information
+// setting the acceleration
+// beginning of the record (lifespan) of the network is index 0
+// acceleration means that snapshots are rarer so the same span can be generated by the journal
+// then update logs can be compressed (toonly one state transition per affected node)
+// epoch, epochcount
+
+type Delta struct {
+	On  int
+	Off int
+}
+
+func oneOutOf(n int) int {
+	t := rand.Intn(n)
+	if t == 0 {
+		return 1
+	}
+	return 0
+}
+
+func deltas(i int) (d []*Delta) {
+	if i == 0 {
+		return []*Delta{
+			&Delta{10, 0},
+			&Delta{20, 0},
+		}
+	}
+	return []*Delta{
+		&Delta{oneOutOf(10), oneOutOf(10)},
+		&Delta{oneOutOf(2), oneOutOf(2)},
+	}
+}
+
+func mockJournalTest(nw *Network, ticker *<-chan time.Time) {
+
+	ids := RandomNodeIDs(100)
+	action := "Off"
+	for n := 0; ; n++ {
+		select {
+		case <-*ticker:
+			var entries []*Entry
+
+			if n == 0 {
+
+				entries = []*Entry{
+					&Entry{
+						Type:   "Node",
+						Action: "On",
+						Object: &SimNode{ID: ids[0]},
+					},
+					&Entry{
+						Type:   "Node",
+						Action: "On",
+						Object: &SimNode{ID: ids[1]},
+					},
+				}
+			} else {
+				sc := &SimConn{
+					Caller: ids[0],
+					Callee: ids[1],
+				}
+				if n%3 == 0 {
+					if action == "On" {
+						action = "Off"
+					} else {
+						action = "On"
+					}
+					entries = append(entries, &Entry{
+						Type:   "Conn",
+						Action: action,
+						Object: sc,
+					})
+				}
+			}
+
+			glog.V(6).Info("entries: %v", entries)
+			nw.AppendEntries(entries...)
+		}
+	}
+}
+
+// MockNetwork generates random connectivity events and posts them
+// to the eventer
+// The journal using the eventer can then be read to visualise or
+// drive connections
+func MockNetwork(eventer *event.TypeMux, ticker <-chan time.Time) {
+	ids := RandomNodeIDs(100)
+	var onNodes []*SimNode
+	offNodes := ids
+	var onConns []*SimConn
+
+	n := 0
+	for _ = range ticker {
+		ds := deltas(n)
+		for i := 0; len(offNodes) > 0 && i < ds[0].On; i++ {
+			c := rand.Intn(len(offNodes))
+			sn := &SimNode{ID: offNodes[c]}
+			err := eventer.Post(&Entry{
+				Type:   "Node",
+				Action: "On",
+				Object: sn,
+			})
+			if err != nil {
+				panic(err.Error())
+			}
+			onNodes = append(onNodes, sn)
+			offNodes = append(offNodes[0:c], offNodes[c+1:]...)
+		}
+		for i := 0; len(onNodes) > 0 && i < ds[0].Off; i++ {
+			c := rand.Intn(len(onNodes))
+			sn := onNodes[c]
+			err := eventer.Post(&Entry{
+				Type:   "Node",
+				Action: "Off",
+				Object: sn,
+			})
+			if err != nil {
+				panic(err.Error())
+			}
+			onNodes = append(onNodes[0:c], onNodes[c+1:]...)
+			offNodes = append(offNodes, sn.ID)
+		}
+		for i := 0; len(onNodes) > 1 && i < ds[1].On; i++ {
+			caller := onNodes[rand.Intn(len(onNodes))].ID
+			callee := onNodes[rand.Intn(len(onNodes))].ID
+			if bytes.Compare(caller[:], callee[:]) >= 0 {
+				i--
+				continue
+			}
+			sc := &SimConn{
+				Caller: caller,
+				Callee: callee,
+			}
+			err := eventer.Post(&Entry{
+				Type:   "Conn",
+				Action: "On",
+				Object: sc,
+			})
+			if err != nil {
+				panic(err.Error())
+			}
+			onConns = append(onConns, sc)
+		}
+		for i := 0; len(onConns) > 0 && i < ds[1].Off; i++ {
+			c := rand.Intn(len(onConns))
+			err := eventer.Post(&Entry{
+				Type:   "Conn",
+				Action: "Off",
+				Object: onConns[c],
+			})
+			if err != nil {
+				panic(err.Error())
+			}
+			onConns = append(onConns[0:c], onConns[c+1:]...)
+		}
+		n++
+	}
+
+}

p2p/simulations/network.go

@@ -1,35 +1,37 @@
 package simulations
 
 import (
-	"bytes"
 	"fmt"
-	"math/rand"
 	"sync"
-	"time"
 
-	"github.com/ethereum/go-ethereum/logger/glog"
 	"github.com/ethereum/go-ethereum/p2p/adapters"
 	"github.com/ethereum/go-ethereum/p2p/discover"
 )
 
 const lablen = 4
 
-func NewNetworkController(n *Network, parent *ResourceController) Controller {
+// func NewMockNetworkController(n *Network, parent *ResourceController) Controller {
+func NewMockNetworkController(journal *Journal, closef func()) Controller {
 	self := NewResourceContoller(
 		&ResourceHandlers{
-			// Destroy: n.Shutdown, nil
+			Destroy: &ResourceHandler{
+				Handle: func(msg interface{}, parent *ResourceController) (interface{}, error) {
+					journal.Close()
+					closef()
+					return nil, nil
+				},
+			},
 			// Create: n.StartNode, NodeConfig
 			// Update: n.Setup, NodeConfig
 			// Retrieve: n.Retrieve,
 			Retrieve: &ResourceHandler{
-				Handle: n.Query,
+				Handle: func(msg interface{}, parent *ResourceController) (interface{}, error) {
+					u := UpdateCy(journal)
+					return u, nil
+				},
 			},
 		},
 	)
-	if parent != nil {
-		parent.SetResource(fmt.Sprintf("%d", parent.id), self)
-	}
-	// self.SetResource("nodes", NewNodesController())
 	return Controller(self)
 }
 
@@ -41,6 +43,8 @@ type Network struct {
 	NodeMap map[discover.NodeID]int
 	Nodes   []*SimNode
 	Journal []*Entry
+	// Journals map[string]*Journal
+	// Events events.Mux
 }
 
 func NewNetwork(m adapters.Messenger) *Network {
@@ -64,15 +68,23 @@ func (self *SimConn) String() string {
 	return fmt.Sprintf("SimConn %v->%v", self.Caller.String()[0:lablen], self.Callee.String()[0:lablen])
 }
 
+// active connections are represented by the SimNode entry object so that
+// you journal updates could filter if passive knowledge about peers is
+// irrelevant
+type SimConn struct {
+	Caller         *discover.NodeID `json:"caller"`
+	Callee         *discover.NodeID `json:"callee"`
+	caller, callee *SimNode
+	// Info
+	// active connection
+	// average throughput, recent average throughput
+}
+
 type NodeConfig struct {
 	ID  *discover.NodeID
 	Run func(adapters.NetAdapter, adapters.Messenger) adapters.ProtoCall
 }
 
-func Key(id []byte) string {
-	return string(id)
-}
-
 func (self *Network) Protocol(id *discover.NodeID) adapters.ProtoCall {
 	self.lock.Lock()
 	defer self.lock.Unlock()
@@ -89,33 +101,6 @@ type QueryConfig struct {
 	Format string // "cy.update", "journal",
 }
 
-type CyData struct {
-	Id     string `json:"id"`
-	Source string `json:"source"`
-	Target string `json:"target"`
-	On     bool   `json:"on"`
-}
-
-type CyElement struct {
-	Data    *CyData `json:"data"`
-	Classes string  `json:"classes"`
-	Group   string  `json:"group"`
-	// selected: false, // whether the element is selected (default false)
-	// selectable: true, // whether the selection state is mutable (default true)
-	// locked: false, // when locked a node's position is immutable (default false)
-	// grabbable: true, // whether the node can be grabbed and moved by the user
-}
-
-type Entry struct {
-	Action string      `json:"action"`
-	Type   string      `json:"type"`
-	Object interface{} `json:"object"`
-}
-
-func (self *Entry) Stirng() string {
-	return fmt.Sprintf("<Action: %v, Type: %v, Data: %v>\n", self.Action, self.Type, self.Object)
-}
-
 func (n *Network) AppendEntries(entries ...*Entry) {
 	n.lock.Lock()
 	n.Journal = append(n.Journal, entries...)
@@ -124,7 +109,7 @@ func (n *Network) AppendEntries(entries ...*Entry) {
 
 type Know struct {
 	Subject *discover.NodeID `json:"subject"`
-	Object  *discover.NodeID `json:"objectr"`
+	Object  *discover.NodeID `json:"object"`
 	// Into
 	// number of attempted connections
 	// time of attempted connections
@@ -133,231 +118,6 @@ type Know struct {
 	// swap balance
 }
 
-// active connections are represented by the SimNode entry object so that
-// you journal updates could filter if passive knowledge about peers is
-// irrelevant
-type SimConn struct {
-	Caller *discover.NodeID `json:"caller"`
-	Callee *discover.NodeID `json:"callee"`
-	// Info
-	// active connection
-	// average throughput, recent average throughput
-}
-
-func (self *Network) CyUpdate() *CyUpdate {
-	self.lock.Lock()
-	defer self.lock.Unlock()
-	added := []*CyElement{}
-	removed := []string{}
-	var el *CyElement
-	for _, entry := range self.Journal {
-		glog.V(6).Infof("journal entry: %v", entry)
-		switch entry.Type {
-		case "Node":
-			el = &CyElement{Group: "nodes", Data: &CyData{Id: entry.Object.(*SimNode).ID.String()[0:lablen]}}
-		case "Conn":
-			// mutually exclusive directed edge (caller -> callee)
-			source := entry.Object.(*SimConn).Caller.String()[0:lablen]
-			target := entry.Object.(*SimConn).Callee.String()[0:lablen]
-			first := source
-			second := target
-			if bytes.Compare([]byte(first), []byte(second)) > 1 {
-				first = target
-				second = source
-			}
-			id := fmt.Sprintf("%v-%v", first, second)
-			el = &CyElement{Group: "edges", Data: &CyData{Id: id, Source: source, Target: target}}
-		case "Know":
-			// independent directed edge (peer0 registers peer1)
-			source := entry.Object.(*Know).Subject.String()[0:lablen]
-			target := entry.Object.(*Know).Object.String()[0:lablen]
-			id := fmt.Sprintf("%v-%v-%v", source, target, "know")
-			el = &CyElement{Group: "edges", Data: &CyData{Id: id, Source: source, Target: target}}
-		}
-		switch entry.Action {
-		case "Add":
-			added = append(added, el)
-		case "Remove":
-			removed = append(removed, el.Data.Id)
-		case "On":
-			el.Data.On = true
-			added = append(added, el)
-		case "Off":
-			el.Data.On = false
-			removed = append(removed, el.Data.Id)
-		}
-	}
-	self.Journal = nil
-	return &CyUpdate{
-		Add:    added,
-		Remove: removed,
-	}
-}
-
-type CyUpdate struct {
-	Add    []*CyElement `json:"add"`
-	Remove []string     `json:"remove"`
-}
-
-func (self *Network) Query(conf interface{}, c *ResourceController) (interface{}, error) {
-	glog.V(6).Infof("query: GET handler ")
-	// config := conf.(*QueryConfig)
-	return interface{}(self.CyUpdate()), nil
-}
-
-// deltas: changes in the number of cumulative actions: non-negative integers.
-// base unit is the fixed minimal interval  between two measurements (time quantum)
-// acceleration : to slow down you just set the base unit higher.
-// to speed up: skip x number of base units
-// frequency: given as the (constant or average) number of base units between measurements
-// if resolution is expressed as the inverse of frequency  = preserved information
-// setting the acceleration
-// beginning of the record (lifespan) of the network is index 0
-// acceleration means that snapshots are rarer so the same span can be generated by the journal
-// then update logs can be compressed (toonly one state transition per affected node)
-// epoch, epochcount
-
-type Delta struct {
-	On  int
-	Off int
-}
-
-func oneOutOf(n int) int {
-	t := rand.Intn(n)
-	if t == 0 {
-		return 1
-	}
-	return 0
-}
-
-func deltas(i int) (d []*Delta) {
-	if i == 0 {
-		return []*Delta{
-			&Delta{10, 0},
-			&Delta{20, 0},
-		}
-	}
-	return []*Delta{
-		&Delta{oneOutOf(10), oneOutOf(10)},
-		&Delta{oneOutOf(2), oneOutOf(2)},
-	}
-}
-
-func mockJournalTest(nw *Network, ticker *<-chan time.Time) {
-
-	ids := RandomNodeIDs(100)
-	action := "Off"
-	for n := 0; ; n++ {
-		select {
-		case <-*ticker:
-			var entries []*Entry
-
-			if n == 0 {
-
-				entries = []*Entry{
-					&Entry{
-						Type:   "Node",
-						Action: "On",
-						Object: &SimNode{ID: ids[0]},
-					},
-					&Entry{
-						Type:   "Node",
-						Action: "On",
-						Object: &SimNode{ID: ids[1]},
-					},
-				}
-			} else {
-				sc := &SimConn{
-					Caller: ids[0],
-					Callee: ids[1],
-				}
-				if n%3 == 0 {
-					if action == "On" {
-						action = "Off"
-					} else {
-						action = "On"
-					}
-					entries = append(entries, &Entry{
-						Type:   "Conn",
-						Action: action,
-						Object: sc,
-					})
-				}
-			}
-
-			glog.V(6).Info("entries: %v", entries)
-			nw.AppendEntries(entries...)
-		}
-	}
-}
-
-func mockJournal(nw *Network, ticker *<-chan time.Time) {
-	ids := RandomNodeIDs(100)
-	var onNodes []*SimNode
-	offNodes := ids
-	var onConns []*SimConn
-
-	for n := 0; ; n++ {
-		select {
-		case <-*ticker:
-			var entries []*Entry
-			ds := deltas(n)
-			for i := 0; len(offNodes) > 0 && i < ds[0].On; i++ {
-				c := rand.Intn(len(offNodes))
-				sn := &SimNode{ID: offNodes[c]}
-				entries = append(entries, &Entry{
-					Type:   "Node",
-					Action: "On",
-					Object: sn,
-				})
-				onNodes = append(onNodes, sn)
-				offNodes = append(offNodes[0:c], offNodes[c+1:]...)
-			}
-			for i := 0; len(onNodes) > 0 && i < ds[0].Off; i++ {
-				c := rand.Intn(len(onNodes))
-				sn := onNodes[c]
-				entries = append(entries, &Entry{
-					Type:   "Node",
-					Action: "Off",
-					Object: sn,
-				})
-				onNodes = append(onNodes[0:c], onNodes[c+1:]...)
-				offNodes = append(offNodes, sn.ID)
-			}
-			for i := 0; len(onNodes) > 1 && i < ds[1].On; i++ {
-				caller := onNodes[rand.Intn(len(onNodes))].ID
-				callee := onNodes[rand.Intn(len(onNodes))].ID
-				if caller == callee {
-					i--
-					continue
-				}
-				sc := &SimConn{
-					Caller: caller,
-					Callee: callee,
-				}
-				entries = append(entries, &Entry{
-					Type:   "Conn",
-					Action: "On",
-					Object: sc,
-				})
-				onConns = append(onConns, sc)
-			}
-			for i := 0; len(onConns) > 0 && i < ds[1].Off; i++ {
-				c := rand.Intn(len(onConns))
-				entries = append(entries, &Entry{
-					Type:   "Conn",
-					Action: "Off",
-					Object: onConns[c],
-				})
-				onConns = append(onConns[0:c], onConns[c+1:]...)
-			}
-			glog.V(6).Info("entries: %v", entries)
-			nw.AppendEntries(entries...)
-		}
-	}
-
-}
-
 func (self *Network) StartNode(conf *NodeConfig) error {
 	self.lock.Lock()
 	defer self.lock.Unlock()

p2p/simulations/session_controller.go

@@ -11,8 +11,8 @@ import (
 	"time"
 
 	"github.com/ethereum/go-ethereum/crypto"
+	"github.com/ethereum/go-ethereum/event"
 	"github.com/ethereum/go-ethereum/logger/glog"
-	"github.com/ethereum/go-ethereum/p2p/adapters"
 	"github.com/ethereum/go-ethereum/p2p/discover"
 )
 
@@ -74,10 +74,17 @@ func NewSessionController() (*ResourceController, chan bool) {
 			Create: &ResourceHandler{
 				Handle: func(msg interface{}, parent *ResourceController) (interface{}, error) {
 					// TODO: take config for type of network
-					network := NewNetwork(&adapters.SimPipe{})
+					eventer := &event.TypeMux{}
 					ticker := time.NewTicker(1000 * time.Millisecond)
-					go mockJournal(network, &ticker.C)
-					return NewNetworkController(network, parent), nil
+					journal := NewJournal(eventer, &Entry{})
+
+					m := NewMockNetworkController(journal, func() { ticker.Stop() })
+					if parent != nil {
+						parent.SetResource(fmt.Sprintf("%d", parent.id), m)
+					}
+					go MockNetwork(eventer, ticker.C)
+					parent.id++
+					return m, nil
 				},
 			},
 

p2p/simulations/session_controller_test.go

@@ -7,6 +7,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/ethereum/go-ethereum/event"
 	"github.com/ethereum/go-ethereum/logger/glog"
 	"github.com/ethereum/go-ethereum/p2p/discover"
 )
@@ -66,9 +67,13 @@ func TestUpdate(t *testing.T) {
 		id := discover.MustHexID(key)
 		ids = append(ids, &id)
 	}
-	network := NewNetwork(nil)
-	NewNetworkController(network, controller)
-	Update(network, ids)
+	eventer := &event.TypeMux{}
+	journal := NewJournal(eventer, &Entry{})
+
+	mc := NewMockNetworkController(journal, nil)
+	controller.SetResource("0", mc)
+	mockNewNodes(eventer, ids)
+	journal.WaitEntries(len(ids))
 	r, err := (&http.Client{}).Do(req)
 	if err != nil {
 		t.Fatalf("unexpected error")
@@ -107,15 +112,12 @@ func TestUpdate(t *testing.T) {
 	}
 }
 
-func Update(self *Network, ids []*discover.NodeID) {
-	self.lock.Lock()
-	defer self.lock.Unlock()
+func mockNewNodes(eventer *event.TypeMux, ids []*discover.NodeID) {
 	for _, id := range ids {
-		e := &Entry{
+		eventer.Post(&Entry{
 			Action: "Add",
 			Type:   "Node",
 			Object: &SimNode{ID: id, config: &NodeConfig{ID: id}},
-		}
-		self.Journal = append(self.Journal, e)
+		})
 	}
 }