p2p/simulations: refactor journal and cytoscape bridge

p2p/simulations/cytoscape.go

@@ -0,0 +1,98 @@
+package simulations
+
+import (
+	"bytes"
+	"fmt"
+
+	"github.com/ethereum/go-ethereum/event"
+	"github.com/ethereum/go-ethereum/logger/glog"
+)
+
+// TODO: to implement cytoscape global behav
+type CyConfig struct {
+}
+
+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)
+}
+
+type CyUpdate struct {
+	Add    []*CyElement `json:"add"`
+	Remove []string     `json:"remove"`
+}
+
+func UpdateCy(conf *CyConfig, j *Journal) (*CyUpdate, error) {
+	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,
+	}, nil
+}

p2p/simulations/journal.go

@@ -0,0 +1,426 @@
+package simulations
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"math/rand"
+	"reflect"
+	"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 {
+	Id      string
+	lock    sync.Mutex
+	counter int
+	cursor  int
+	quitc   chan bool
+	Events  []*event.Event
+}
+
+// NewJournal constructor
+// Journal can get input events from subscriptions, add event logs
+// or scheduled replay of events from another journal
+//
+// see the Read and TimedRead iterators for use
+// the Journal is safe for concurrent reads and writes
+func NewJournal() *Journal {
+	return &Journal{quitc: make(chan bool)}
+}
+
+// Subscribe takes an event.TypeMux and subscibes to types
+// and launches a gorourine that appends any new event to the event log
+// used for journalling history of a network
+// the goroutine terminates when the journal is closed
+func (self *Journal) Subscribe(eventer *event.TypeMux, types ...interface{}) {
+	glog.V(6).Infof("subscribe")
+	sub := eventer.Subscribe(types...)
+	go func() {
+		defer sub.Unsubscribe()
+		for {
+			select {
+			case ev := <-sub.Chan():
+				glog.V(6).Infof("appebd ev %v", ev)
+				self.append(ev)
+			case <-self.quitc:
+				return
+			}
+		}
+	}()
+}
+
+// AddJournal appends the event log of another journal to the receiver's one
+func (self *Journal) AddJournal(j *Journal) {
+	self.append(j.Events...)
+}
+
+// NewJournalFromJSON decodes a JSON serialised events log
+// into a journal struct
+// used to replay recorded history
+func NewJournalFromJSON(b []byte) (*Journal, error) {
+	self := NewJournal()
+	err := json.Unmarshal(b, self)
+	if err != nil {
+		return nil, err
+	}
+	return self, nil
+}
+
+// Replay replays the events of another journal preserving (relative) timing of events
+// params:
+// * acc: using acceleration factor acc
+// * journal: journal to use
+// * eventer: where to post the replayed events
+func Replay(acc float64, j *Journal, eventer *event.TypeMux) {
+	f := func(d interface{}) bool {
+		// reposts the data with the eventer (the data receives a new timestamp)
+		eventer.Post(d)
+		return true
+	}
+	j.TimedRead(acc, f)
+}
+
+// Snapshot creates a snapshot out of the journal
+// this is simply done by reading the event log backwards and mark the last action
+// on a node/connection ignoring all earlier mentions
+// TODO: implmented
+func Snapshot(conf *SnapshotConfig, j *Journal) (*Journal, error) {
+	return nil, fmt.Errorf("snapshot not implemented")
+}
+
+func (self *Journal) Close() {
+	close(self.quitc)
+}
+
+func (self *Journal) append(evs ...*event.Event) {
+	self.lock.Lock()
+	defer self.lock.Unlock()
+	self.Events = append(self.Events, evs...)
+	self.counter++
+}
+
+func (self *Journal) NewEntries() int {
+	self.lock.Lock()
+	defer self.lock.Unlock()
+	return self.counter - self.cursor
+}
+
+func (self *Journal) WaitEntries(n int) {
+	for self.NewEntries() < n {
+		glog.V(6).Infof(".")
+	}
+}
+
+func (self *Journal) Read(f func(*event.Event) bool) (read int, err error) {
+	self.lock.Lock()
+	defer self.lock.Unlock()
+	glog.V(6).Infof("read out of %v", len(self.Events))
+	ok := true
+	for self.cursor < len(self.Events) && ok {
+		glog.V(6).Infof("read %v", read)
+		read++
+		ok = f(self.Events[self.cursor])
+		self.cursor++
+		select {
+		case <-self.quitc:
+			break
+		default:
+		}
+	}
+	self.reset(self.cursor)
+	return read, nil
+}
+
+// TimedRead reads the events but blocks for intervals that correspond to
+// the original time intervals,
+// NOTE: the events' timestamps are supposed to be strictly ordered otherwise
+// the call panics.
+// acc is an acceleration factor
+func (self *Journal) TimedRead(acc float64, f func(interface{}) bool) (read int, err error) {
+	var lastEvent time.Time
+	timer := time.NewTimer(0)
+	var data interface{}
+	h := func(ev *event.Event) bool {
+		// wait for the interval time passes event time
+		if ev.Time.Before(lastEvent) {
+			panic("events not ordered")
+		}
+		interval := ev.Time.Sub(lastEvent)
+		glog.V(6).Infof("reset timer to interval %v", interval)
+		timer.Reset(time.Duration(acc) * interval)
+		lastEvent = ev.Time
+		data = ev.Data
+		return false
+	}
+	var n int
+	for {
+		// Read blocks for the iteration. need to read one event at a time so that
+		// waiting for the timer to go off does not block concurrent access to the journal
+		n, err = self.Read(h)
+		if read > 0 {
+			select {
+			case <-self.quitc:
+				break
+			case <-timer.C:
+			}
+		}
+		read += n
+		if err != nil || n == 0 || !f(data) {
+			break
+		}
+	}
+	return read, err
+}
+
+func (self *Journal) Reset(n int) {
+	self.lock.Lock()
+	defer self.lock.Unlock()
+	self.reset(n)
+}
+
+func (self *Journal) reset(n int) {
+	length := len(self.Events)
+	if length == 0 {
+		return
+	}
+	if n >= length-1 {
+		n = length - 1
+	}
+	glog.V(6).Infof("cursor reset from %v to %v/%v (%v)", self.cursor, n, len(self.Events), 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
+}
+
+type SnapshotConfig struct {
+	Id string
+}
+
+type JournalPlayConfig struct {
+	Id      string
+	SpeedUp float64
+	Journal *Journal
+	Events  []string
+}
+
+func NewJournalPlayersController(eventer *event.TypeMux) Controller {
+	self := NewResourceContoller(
+		&ResourceHandlers{
+			// POST /o/players/
+			Create: &ResourceHandler{
+				Handle: func(msg interface{}, parent *ResourceController) (interface{}, error) {
+					conf := msg.(*JournalPlayConfig)
+					go Replay(conf.SpeedUp, conf.Journal, eventer)
+					c := NewJournalPlayerController(conf)
+					parent.SetResource(conf.Id, c)
+					return nil, nil
+				},
+				Type: reflect.TypeOf(&JournalPlayConfig{}),
+			},
+		})
+	return self
+}
+
+func NewJournalPlayerController(conf *JournalPlayConfig) Controller {
+	self := NewResourceContoller(
+		&ResourceHandlers{
+			// GET /0/players/<playerId>
+			Retrieve: &ResourceHandler{
+				Handle: func(msg interface{}, parent *ResourceController) (interface{}, error) {
+					return nil, fmt.Errorf("info about journal player not implemented")
+				},
+			},
+			// DELETE /0/players/<playerId>
+			Destroy: &ResourceHandler{
+				Handle: func(msg interface{}, parent *ResourceController) (interface{}, error) {
+					conf.Journal.Close() // terminate Replay-> TimedRead routine
+					parent.DeleteResource(conf.Id)
+					return nil, nil
+				},
+			},
+		})
+	return self
+}
+
+type MockerConfig struct {
+	// TODO: frequency/volume etc.
+	Id             string
+	UpdateInterval time.Duration
+}
+
+func NewMockersController(eventer *event.TypeMux) Controller {
+	self := NewResourceContoller(
+		&ResourceHandlers{
+			// Create: n.StartNode, NodeConfig
+			Create: &ResourceHandler{
+				Handle: func(msg interface{}, parent *ResourceController) (interface{}, error) {
+					conf := msg.(*MockerConfig)
+					ticker := time.NewTicker(conf.UpdateInterval)
+					go MockEvents(eventer, ticker.C)
+					c := NewMockerController(conf, ticker)
+					parent.SetResource(conf.Id, c)
+					return nil, nil
+				},
+				Type: reflect.TypeOf(&MockerConfig{}),
+			},
+		})
+	return self
+}
+
+func NewMockerController(conf *MockerConfig, ticker *time.Ticker) Controller {
+	self := NewResourceContoller(
+		&ResourceHandlers{
+			// GET /0/mockevents/<mockerId>
+			Retrieve: &ResourceHandler{
+				Handle: func(msg interface{}, parent *ResourceController) (interface{}, error) {
+					return nil, fmt.Errorf("info about mocker not implemented")
+				},
+			},
+			// DELETE /0/mockevents/<mockerId>
+			Destroy: &ResourceHandler{
+				Handle: func(msg interface{}, parent *ResourceController) (interface{}, error) {
+					ticker.Stop() //terminate MockEvents routine
+					parent.DeleteResource(conf.Id)
+					return nil, nil
+				},
+			},
+		})
+	return self
+}
+
+// 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)},
+	}
+}
+
+// MockEvents 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 MockEvents(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/journal_test.go

@@ -0,0 +1,72 @@
+package simulations
+
+import (
+	// "encoding/json"
+	"testing"
+	"time"
+
+	"github.com/ethereum/go-ethereum/event"
+	// "github.com/ethereum/go-ethereum/p2p/discover"
+	// "github.com/ethereum/go-ethereum/logger/glog"
+)
+
+func testEvents(intervals ...int) (events []*event.Event) {
+	t := time.Now()
+	for i, interval := range intervals {
+		t = t.Add(time.Duration(interval) * time.Millisecond)
+		events = append(events, &event.Event{
+			Time: t,
+			Data: interface{}(&Entry{
+				Type:   "node",
+				Action: "Off",
+				Object: interface{}(i),
+			}),
+		})
+	}
+	return events
+}
+
+func TestTimedRead(t *testing.T) {
+	// keys := []string{
+	// 	"aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80",
+	// 	"f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3",
+	// }
+	// var ids []*discover.NodeID
+	// for _, key := range keys {
+	// 	id := discover.MustHexID(key)
+	// 	ids = append(ids, &id)
+	// }
+
+	j := NewJournal()
+	intervals := []int{100, 200, 300, 300, 100, 200}
+	j.Events = testEvents(intervals...)
+	var newTimes []time.Time
+	var i int
+	acc := 0.5
+	length := 4
+	f := func(data interface{}) bool {
+		_ = data.(*Entry)
+		newTimes = append(newTimes, time.Now())
+		i++
+		return i <= length
+	}
+	start := time.Now()
+	read, err := j.TimedRead(acc, f)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if read != 5 {
+		t.Fatalf("incorrect number of events read: expected 5, got %v", read)
+	}
+	for i, ti := range newTimes {
+		expInt := time.Duration(acc*float64(intervals[i])) * time.Millisecond
+		gotInt := ti.Sub(start)
+		if gotInt-expInt > 1*time.Millisecond {
+			t.Fatalf("journal timed read incorrect interval: expected %v ,got %v", expInt, gotInt)
+		}
+		start = ti
+	}
+}
+
+// func TestReplay(t *testing.T) {
+// }

p2p/simulations/network.go

@@ -1,12 +1,11 @@
 package simulations
 
 import (
-	"bytes"
 	"fmt"
-	"math/rand"
+	"reflect"
 	"sync"
-	"time"
 
+	"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"
@@ -14,22 +13,48 @@ import (
 
 const lablen = 4
 
-func NewNetworkController(n *Network, parent *ResourceController) Controller {
+type NetworkConfig struct {
+	// Type   NetworkType
+	// Config json.RawMessage // type-specific configs
+	// type
+	// Events []string
+	Id string
+}
+
+func NewNetworkController(conf *NetworkConfig, eventer *event.TypeMux, journal *Journal) Controller {
 	self := NewResourceContoller(
 		&ResourceHandlers{
-			// Destroy: n.Shutdown, nil
+			// GET /<networkId>/
-			// Create: n.StartNode, NodeConfig
-			// Update: n.Setup, NodeConfig
-			// Retrieve: n.Retrieve,
 			Retrieve: &ResourceHandler{
-				Handle: n.Query,
+				Handle: func(msg interface{}, parent *ResourceController) (interface{}, error) {
+					glog.V(6).Infof("msg: %v", msg)
+					cyConfig, ok := msg.(*CyConfig)
+					if ok {
+						return UpdateCy(cyConfig, journal)
+					}
+					snapshotConfig, ok := msg.(*SnapshotConfig)
+					if ok {
+						return Snapshot(snapshotConfig, journal)
+					}
+					return nil, fmt.Errorf("invalid json body: must be CyConfig or SnapshotConfig")
+				},
+				Type: reflect.TypeOf(&CyConfig{}),
+			},
+			// DELETE /<networkId>/
+			Destroy: &ResourceHandler{
+				Handle: func(msg interface{}, parent *ResourceController) (interface{}, error) {
+					parent.DeleteResource(conf.Id)
+					return nil, nil
+				},
 			},
 		},
 	)
-	if parent != nil {
+	// subscribe to all event entries (generated)
-		parent.SetResource(fmt.Sprintf("%d", parent.id), self)
+	journal.Subscribe(eventer, &Entry{})
-	}
+	// self.SetResource("nodes", NewNodesController(eventer))
-	// self.SetResource("nodes", NewNodesController())
+	// self.SetResource("connections", NewConnectionsController(eventer))
+	self.SetResource("mockevents", NewMockersController(eventer))
+	self.SetResource("journals", NewJournalPlayersController(eventer))
 	return Controller(self)
 }
 
@@ -40,7 +65,6 @@ type Network struct {
 	lock    sync.RWMutex
 	NodeMap map[discover.NodeID]int
 	Nodes   []*SimNode
-	Journal []*Entry
 }
 
 func NewNetwork(m adapters.Messenger) *Network {
@@ -64,15 +88,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,42 +121,9 @@ 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...)
-	n.lock.Unlock()
-}
-
 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 +132,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()
@@ -372,7 +146,7 @@ func (self *Network) StartNode(conf *NodeConfig) error {
 		simnet.Run = conf.Run(simnet, self.Messenger)
 	}
 	self.NodeMap[*id] = len(self.Nodes)
-	self.Nodes = append(self.Nodes, &SimNode{id, conf, adapters.NetAdapter(simnet)})
+	self.Nodes = append(self.Nodes, &SimNode{id, conf, simnet})
 	return nil
 }
 

p2p/simulations/session_controller.go

@@ -8,11 +8,10 @@ import (
 	"io/ioutil"
 	"reflect"
 	"sync"
-	"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"
 )
 
@@ -73,12 +72,18 @@ func NewSessionController() (*ResourceController, chan bool) {
 
 			Create: &ResourceHandler{
 				Handle: func(msg interface{}, parent *ResourceController) (interface{}, error) {
-					// TODO: take config for type of network
+					conf := msg.(*NetworkConfig)
-					network := NewNetwork(&adapters.SimPipe{})
+					m := NewNetworkController(conf, &event.TypeMux{}, NewJournal())
-					ticker := time.NewTicker(1000 * time.Millisecond)
+					if len(conf.Id) == 0 {
-					go mockJournal(network, &ticker.C)
+						conf.Id = fmt.Sprintf("%d", parent.id)
-					return NewNetworkController(network, parent), nil
+					}
+					if parent != nil {
+						parent.SetResource(conf.Id, m)
+					}
+					parent.id++
+					return m, nil
 				},
+				Type: reflect.TypeOf(&NetworkConfig{}),
 			},
 
 			Destroy: &ResourceHandler{
@@ -149,6 +154,10 @@ func (self *ResourceController) SetResource(id string, c Controller) {
 	}
 }
 
+func (self *ResourceController) DeleteResource(id string) {
+	delete(self.controllers, id)
+}
+
 func RandomNodeID() *discover.NodeID {
 	key, err := crypto.GenerateKey()
 	if err != nil {

p2p/simulations/session_controller_test.go

@@ -1,12 +1,14 @@
 package simulations
 
 import (
+	"bytes"
 	"fmt"
 	"io/ioutil"
 	"net/http"
 	"testing"
 	"time"
 
+	"github.com/ethereum/go-ethereum/event"
 	"github.com/ethereum/go-ethereum/logger/glog"
 	"github.com/ethereum/go-ethereum/p2p/discover"
 )
@@ -32,7 +34,6 @@ func url(port, path string) string {
 
 func TestQuit(t *testing.T) {
 	req, err := http.NewRequest("DELETE", url(port, ""), nil)
-	// req, err := http.NewRequest("PUT", url(""), nil)
 	if err != nil {
 		t.Fatalf("unexpected error")
 	}
@@ -53,10 +54,7 @@ func TestQuit(t *testing.T) {
 }
 
 func TestUpdate(t *testing.T) {
-	req, err := http.NewRequest("GET", url(port, "0"), nil)
+
-	if err != nil {
-		t.Fatalf("unexpected error")
-	}
 	keys := []string{
 		"aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80aa7cca80",
 		"f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3f5ae22c3",
@@ -66,12 +64,25 @@ func TestUpdate(t *testing.T) {
 		id := discover.MustHexID(key)
 		ids = append(ids, &id)
 	}
-	network := NewNetwork(nil)
+
-	NewNetworkController(network, controller)
+	eventer := &event.TypeMux{}
-	Update(network, ids)
+	journal := NewJournal()
+	journal.Subscribe(eventer, &Entry{})
+	mockNewNodes(eventer, ids)
+	conf := &NetworkConfig{
+		Id: "0",
+	}
+	mc := NewNetworkController(conf, eventer, journal)
+	controller.SetResource(conf.Id, mc)
+	journal.WaitEntries(len(ids))
+
+	req, err := http.NewRequest("GET", url(port, "0"), bytes.NewReader([]byte("{}")))
+	if err != nil {
+		t.Fatalf("unexpected error creating request: %v", err)
+	}
 	r, err := (&http.Client{}).Do(req)
 	if err != nil {
-		t.Fatalf("unexpected error")
+		t.Fatalf("unexpected error on http.Client request: %v", err)
 	}
 	resp, err := ioutil.ReadAll(r.Body)
 	if err != nil {
@@ -107,15 +118,18 @@ func TestUpdate(t *testing.T) {
 	}
 }
 
-func Update(self *Network, ids []*discover.NodeID) {
+func mockNewNodes(eventer *event.TypeMux, ids []*discover.NodeID) {
-	self.lock.Lock()
+	glog.V(6).Infof("mock starting")
-	defer self.lock.Unlock()
 	for _, id := range ids {
-		e := &Entry{
+		glog.V(6).Infof("mock adding node %v", id)
+		eventer.Post(&Entry{
 			Action: "Add",
 			Type:   "Node",
 			Object: &SimNode{ID: id, config: &NodeConfig{ID: id}},
-		}
+		})
-		self.Journal = append(self.Journal, e)
 	}
 }
+
+// func TestReplay(t *testing.T) {
+
+// }