diff options
| -rw-r--r-- | block_pool.go | 5 | ||||
| -rw-r--r-- | ethchain/state_transition.go | 14 | ||||
| -rw-r--r-- | peer.go | 2 | ||||
| -rw-r--r-- | websocket/client.go | 122 | ||||
| -rw-r--r-- | websocket/message.go | 14 | ||||
| -rw-r--r-- | websocket/server.go | 127 |
6 files changed, 272 insertions, 12 deletions
diff --git a/block_pool.go b/block_pool.go index 957b7601b..003d1db58 100644 --- a/block_pool.go +++ b/block_pool.go @@ -11,7 +11,6 @@ import ( "github.com/ethereum/eth-go/ethchain" "github.com/ethereum/eth-go/ethlog" "github.com/ethereum/eth-go/ethutil" - "github.com/ethereum/eth-go/ethwire" ) var poollogger = ethlog.NewLogger("BPOOL") @@ -99,8 +98,8 @@ func (self *BlockPool) Add(b *ethchain.Block, peer *Peer) { self.pool[hash] = &block{peer, peer, b, time.Now(), 0} if !self.eth.BlockChain().HasBlock(b.PrevHash) && !self.fetchingHashes { - poollogger.Infof("Unknown block, requesting parent (%x...)\n", b.PrevHash[0:4]) - peer.QueueMessage(ethwire.NewMessage(ethwire.MsgGetBlockHashesTy, []interface{}{b.PrevHash, uint32(256)})) + //poollogger.Infof("Unknown block, requesting parent (%x...)\n", b.PrevHash[0:4]) + //peer.QueueMessage(ethwire.NewMessage(ethwire.MsgGetBlockHashesTy, []interface{}{b.PrevHash, uint32(256)})) } } else if self.pool[hash] != nil { self.pool[hash].block = b diff --git a/ethchain/state_transition.go b/ethchain/state_transition.go index c1180a641..096464963 100644 --- a/ethchain/state_transition.go +++ b/ethchain/state_transition.go @@ -142,14 +142,12 @@ func (self *StateTransition) preCheck() (err error) { func (self *StateTransition) TransitionState() (err error) { statelogger.Debugf("(~) %x\n", self.tx.Hash()) - /* - defer func() { - if r := recover(); r != nil { - logger.Infoln(r) - err = fmt.Errorf("state transition err %v", r) - } - }() - */ + defer func() { + if r := recover(); r != nil { + statelogger.Infoln(r) + err = fmt.Errorf("state transition err %v", r) + } + }() // XXX Transactions after this point are considered valid. if err = self.preCheck(); err != nil { @@ -320,7 +320,7 @@ out: case msg := <-p.outputQueue: if !p.statusKnown { switch msg.Type { - case ethwire.MsgStatusTy, ethwire.MsgGetTxsTy, ethwire.MsgTxTy, ethwire.MsgGetBlockHashesTy, ethwire.MsgBlockHashesTy, ethwire.MsgGetBlocksTy, ethwire.MsgBlockTy: + case ethwire.MsgGetTxsTy, ethwire.MsgTxTy, ethwire.MsgGetBlockHashesTy, ethwire.MsgBlockHashesTy, ethwire.MsgGetBlocksTy, ethwire.MsgBlockTy: break skip } } diff --git a/websocket/client.go b/websocket/client.go new file mode 100644 index 000000000..1ff0d3f64 --- /dev/null +++ b/websocket/client.go @@ -0,0 +1,122 @@ +package websocket + +import ( + "fmt" + "io" + + ws "code.google.com/p/go.net/websocket" +) + +const channelBufSize = 100 + +var maxId int = 0 + +type MsgFunc func(c *Client, msg *Message) + +// Chat client. +type Client struct { + id int + ws *ws.Conn + server *Server + ch chan *Message + doneCh chan bool + + onMessage MsgFunc +} + +// Create new chat client. +func NewClient(ws *ws.Conn, server *Server) *Client { + + if ws == nil { + panic("ws cannot be nil") + } + + if server == nil { + panic("server cannot be nil") + } + + maxId++ + ch := make(chan *Message, channelBufSize) + doneCh := make(chan bool) + + return &Client{maxId, ws, server, ch, doneCh, nil} +} + +func (c *Client) Id() int { + return c.id +} + +func (c *Client) Conn() *ws.Conn { + return c.ws +} + +func (c *Client) Write(data interface{}, seed int) { + msg := &Message{Seed: seed, Data: data} + select { + case c.ch <- msg: + default: + c.server.Del(c) + err := fmt.Errorf("client %d is disconnected.", c.id) + c.server.Err(err) + } +} + +func (c *Client) Done() { + c.doneCh <- true +} + +// Listen Write and Read request via chanel +func (c *Client) Listen() { + go c.listenWrite() + c.listenRead() +} + +// Listen write request via chanel +func (c *Client) listenWrite() { + logger.Debugln("Listening write to client") + for { + select { + + // send message to the client + case msg := <-c.ch: + logger.Debugln("Send:", msg) + ws.JSON.Send(c.ws, msg) + + // receive done request + case <-c.doneCh: + c.server.Del(c) + c.doneCh <- true // for listenRead method + return + } + } +} + +// Listen read request via chanel +func (c *Client) listenRead() { + logger.Debugln("Listening read from client") + for { + select { + + // receive done request + case <-c.doneCh: + c.server.Del(c) + c.doneCh <- true // for listenWrite method + return + + // read data from ws connection + default: + var msg Message + err := ws.JSON.Receive(c.ws, &msg) + if err == io.EOF { + c.doneCh <- true + } else if err != nil { + c.server.Err(err) + } else { + logger.Debugln(&msg) + if c.onMessage != nil { + c.onMessage(c, &msg) + } + } + } + } +} diff --git a/websocket/message.go b/websocket/message.go new file mode 100644 index 000000000..30e284814 --- /dev/null +++ b/websocket/message.go @@ -0,0 +1,14 @@ +package websocket + +import "github.com/ethereum/eth-go/ethutil" + +type Message struct { + Call string `json:"call"` + Args []interface{} `json:"args"` + Seed int `json:"seed"` + Data interface{} `json:"data"` +} + +func (self *Message) Arguments() *ethutil.Value { + return ethutil.NewValue(self.Args) +} diff --git a/websocket/server.go b/websocket/server.go new file mode 100644 index 000000000..648f3fdc3 --- /dev/null +++ b/websocket/server.go @@ -0,0 +1,127 @@ +package websocket + +import ( + "net/http" + + "github.com/ethereum/eth-go/ethlog" + + ws "code.google.com/p/go.net/websocket" +) + +var logger = ethlog.NewLogger("WS") + +// Chat server. +type Server struct { + httpServ string + pattern string + messages []*Message + clients map[int]*Client + addCh chan *Client + delCh chan *Client + sendAllCh chan string + doneCh chan bool + errCh chan error + msgFunc MsgFunc +} + +// Create new chat server. +func NewServer(pattern, httpServ string) *Server { + clients := make(map[int]*Client) + addCh := make(chan *Client) + delCh := make(chan *Client) + sendAllCh := make(chan string) + doneCh := make(chan bool) + errCh := make(chan error) + + return &Server{ + httpServ, + pattern, + nil, + clients, + addCh, + delCh, + sendAllCh, + doneCh, + errCh, + nil, + } +} + +func (s *Server) Add(c *Client) { + s.addCh <- c +} + +func (s *Server) Del(c *Client) { + s.delCh <- c +} + +func (s *Server) SendAll(msg string) { + s.sendAllCh <- msg +} + +func (s *Server) Done() { + s.doneCh <- true +} + +func (s *Server) Err(err error) { + s.errCh <- err +} + +func (s *Server) servHTTP() { + logger.Debugln("Serving http", s.httpServ) + err := http.ListenAndServe(s.httpServ, nil) + + logger.Warnln(err) +} + +func (s *Server) MessageFunc(f MsgFunc) { + s.msgFunc = f +} + +// Listen and serve. +// It serves client connection and broadcast request. +func (s *Server) Listen() { + logger.Debugln("Listening server...") + + // ws handler + onConnected := func(ws *ws.Conn) { + defer func() { + err := ws.Close() + if err != nil { + s.errCh <- err + } + }() + + client := NewClient(ws, s) + client.onMessage = s.msgFunc + s.Add(client) + client.Listen() + } + // Disable Origin check. Request don't need to come necessarily from origin. + http.HandleFunc(s.pattern, func(w http.ResponseWriter, req *http.Request) { + s := ws.Server{Handler: ws.Handler(onConnected)} + s.ServeHTTP(w, req) + }) + logger.Debugln("Created handler") + + go s.servHTTP() + + for { + select { + + // Add new a client + case c := <-s.addCh: + s.clients[c.id] = c + + // del a client + case c := <-s.delCh: + delete(s.clients, c.id) + + case err := <-s.errCh: + logger.Debugln("Error:", err.Error()) + + case <-s.doneCh: + return + } + } +} |