Merge pull request #1449 from Bytom/dev

Dev

api/api.go

@@ -23,6 +23,7 @@ import (
 	"github.com/bytom/net/http/gzip"
 	"github.com/bytom/net/http/httpjson"
 	"github.com/bytom/net/http/static"
+	"github.com/bytom/net/websocket"
 	"github.com/bytom/netsync"
 	"github.com/bytom/protocol"
 	"github.com/bytom/protocol/bc"
@@ -104,17 +105,17 @@ func (wh *waitHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 
 // API is the scheduling center for server
 type API struct {
-	sync          *netsync.SyncManager
-	wallet        *wallet.Wallet
-	accessTokens  *accesstoken.CredentialStore
-	chain         *protocol.Chain
-	server        *http.Server
-	handler       http.Handler
-	txFeedTracker *txfeed.Tracker
-	cpuMiner      *cpuminer.CPUMiner
-	miningPool    *miningpool.MiningPool
-
-	newBlockCh chan *bc.Hash
+	sync            *netsync.SyncManager
+	wallet          *wallet.Wallet
+	accessTokens    *accesstoken.CredentialStore
+	chain           *protocol.Chain
+	server          *http.Server
+	handler         http.Handler
+	txFeedTracker   *txfeed.Tracker
+	cpuMiner        *cpuminer.CPUMiner
+	miningPool      *miningpool.MiningPool
+	notificationMgr *websocket.WSNotificationManager
+	newBlockCh      chan *bc.Hash
 }
 
 func (a *API) initServer(config *cfg.Config) {
@@ -168,7 +169,7 @@ func (a *API) StartServer(address string) {
 }
 
 // NewAPI create and initialize the API
-func NewAPI(sync *netsync.SyncManager, wallet *wallet.Wallet, txfeeds *txfeed.Tracker, cpuMiner *cpuminer.CPUMiner, miningPool *miningpool.MiningPool, chain *protocol.Chain, config *cfg.Config, token *accesstoken.CredentialStore, newBlockCh chan *bc.Hash) *API {
+func NewAPI(sync *netsync.SyncManager, wallet *wallet.Wallet, txfeeds *txfeed.Tracker, cpuMiner *cpuminer.CPUMiner, miningPool *miningpool.MiningPool, chain *protocol.Chain, config *cfg.Config, token *accesstoken.CredentialStore, newBlockCh chan *bc.Hash, notificationMgr *websocket.WSNotificationManager) *API {
 	api := &API{
 		sync:          sync,
 		wallet:        wallet,
@@ -178,7 +179,8 @@ func NewAPI(sync *netsync.SyncManager, wallet *wallet.Wallet, txfeeds *txfeed.Tr
 		cpuMiner:      cpuMiner,
 		miningPool:    miningPool,
 
-		newBlockCh: newBlockCh,
+		newBlockCh:      newBlockCh,
+		notificationMgr: notificationMgr,
 	}
 	api.buildHandler()
 	api.initServer(config)
@@ -296,6 +298,8 @@ func (a *API) buildHandler() {
 
 	m.Handle("/get-merkle-proof", jsonHandler(a.getMerkleProof))
 
+	m.HandleFunc("/websocket-subscribe", a.websocketHandler)
+
 	handler := latencyHandler(m, walletEnable)
 	handler = webAssetsHandler(handler)
 	handler = gzip.Handler{Handler: handler}

api/websocket.go

@@ -0,0 +1,32 @@
+package api
+
+import (
+	"net/http"
+	"time"
+
+	log "github.com/sirupsen/logrus"
+
+	"github.com/bytom/net/websocket"
+)
+
+// timeZeroVal is simply the zero value for a time.Time and is used to avoid
+// creating multiple instances.
+var timeZeroVal time.Time
+
+// WebsocketHandler handles connections and requests from websocket client
+func (a *API) websocketHandler(w http.ResponseWriter, r *http.Request) {
+	log.WithField("remoteAddress", r.RemoteAddr).Info("New websocket client")
+
+	client, err := websocket.NewWebsocketClient(w, r, a.notificationMgr)
+	if err != nil {
+		log.WithField("error", err).Error("Failed to new websocket client")
+		http.Error(w, "400 Bad Request.", http.StatusBadRequest)
+		return
+	}
+
+	a.notificationMgr.AddClient(client)
+	client.Start()
+	client.WaitForShutdown()
+	a.notificationMgr.RemoveClient(client)
+	log.WithField("address", r.RemoteAddr).Infoln("Disconnected websocket client")
+}

cmd/bytomd/commands/run_node.go

@@ -44,6 +44,10 @@ func init() {
 	// log flags
 	runNodeCmd.Flags().String("log_file", config.LogFile, "Log output file")
 
+	// websocket flags
+	runNodeCmd.Flags().Int("ws.max_num_websockets", config.Websocket.MaxNumWebsockets, "Max number of websocket connections")
+	runNodeCmd.Flags().Int("ws.max_num_concurrent_reqs", config.Websocket.MaxNumConcurrentReqs, "Max number of concurrent websocket requests that may be processed concurrently")
+
 	RootCmd.AddCommand(runNodeCmd)
 }
 

config/config.go

@@ -18,11 +18,12 @@ type Config struct {
 	// Top level options use an anonymous struct
 	BaseConfig `mapstructure:",squash"`
 	// Options for services
-	P2P    *P2PConfig     `mapstructure:"p2p"`
-	Wallet *WalletConfig  `mapstructure:"wallet"`
-	Auth   *RPCAuthConfig `mapstructure:"auth"`
-	Web    *WebConfig     `mapstructure:"web"`
-	Simd   *SimdConfig    `mapstructure:"simd"`
+	P2P       *P2PConfig       `mapstructure:"p2p"`
+	Wallet    *WalletConfig    `mapstructure:"wallet"`
+	Auth      *RPCAuthConfig   `mapstructure:"auth"`
+	Web       *WebConfig       `mapstructure:"web"`
+	Simd      *SimdConfig      `mapstructure:"simd"`
+	Websocket *WebsocketConfig `mapstructure:"ws"`
 }
 
 // Default configurable parameters.
@@ -34,6 +35,7 @@ func DefaultConfig() *Config {
 		Auth:       DefaultRPCAuthConfig(),
 		Web:        DefaultWebConfig(),
 		Simd:       DefaultSimdConfig(),
+		Websocket:  DefaultWebsocketConfig(),
 	}
 }
 
@@ -141,6 +143,11 @@ type SimdConfig struct {
 	Enable bool `mapstructure:"enable"`
 }
 
+type WebsocketConfig struct {
+	MaxNumWebsockets     int `mapstructure:"max_num_websockets"`
+	MaxNumConcurrentReqs int `mapstructure:"max_num_concurrent_reqs"`
+}
+
 // Default configurable rpc's auth parameters.
 func DefaultRPCAuthConfig() *RPCAuthConfig {
 	return &RPCAuthConfig{
@@ -171,6 +178,13 @@ func DefaultSimdConfig() *SimdConfig {
 	}
 }
 
+func DefaultWebsocketConfig() *WebsocketConfig {
+	return &WebsocketConfig{
+		MaxNumWebsockets:     25,
+		MaxNumConcurrentReqs: 20,
+	}
+}
+
 //-----------------------------------------------------------------------------
 // Utils
 

consensus/general.go

@@ -129,6 +129,7 @@ var TestNetParams = Params{
 	Checkpoints: []Checkpoint{
 		{10303, bc.NewHash([32]byte{0x3e, 0x94, 0x5d, 0x35, 0x70, 0x30, 0xd4, 0x3b, 0x3d, 0xe3, 0xdd, 0x80, 0x67, 0x29, 0x9a, 0x5e, 0x09, 0xf9, 0xfb, 0x2b, 0xad, 0x5f, 0x92, 0xc8, 0x69, 0xd1, 0x42, 0x39, 0x74, 0x9a, 0xd1, 0x1c})},
 		{40000, bc.NewHash([32]byte{0x6b, 0x13, 0x9a, 0x5b, 0x76, 0x77, 0x9b, 0xd4, 0x1c, 0xec, 0x53, 0x68, 0x44, 0xbf, 0xf4, 0x48, 0x94, 0x3d, 0x16, 0xe3, 0x9b, 0x2e, 0xe8, 0xa1, 0x0f, 0xa0, 0xbc, 0x7d, 0x2b, 0x17, 0x55, 0xfc})},
+		{78000, bc.NewHash([32]byte{0xa9, 0x03, 0xc0, 0x0c, 0x62, 0x1a, 0x3d, 0x00, 0x7f, 0xd8, 0x5d, 0x51, 0xba, 0x43, 0xe4, 0xd0, 0xe3, 0xc5, 0xd4, 0x8f, 0x30, 0xb5, 0x5f, 0xa5, 0x77, 0x62, 0xd8, 0x8b, 0x11, 0x81, 0x5f, 0xb4})},
 	},
 }