8fb3db8136
* Move and rename BlockchainCodec, BlockchainCodecRequest. These are not specifically "blockchain", rather they are specific to how gorilla/rpc works. * Move claimtrie-related service/handlers to jsonrpc_claimtrie.go. * Pull out decode logic into named func newBlockHeaderElectrum(). * Rename BlockchainService -> BlockchainBlockService. * Drop http.Request arg from handlers, and use RegisterTCPService(). * Implement GetStatus() to pull data from HashXStatus table. * Make the service objects independent, so we don't have inheritance. * Add core session/subscription logic (session.go). Implement subsribe/unsubscribe handlers. * Support both pure JSON and JSON-over-HTTP services. Forward NotifierChan messages to sessionManager. * Only assign default port (50001) if neither --json-rpc-port nor --json-rpc-http-port are specified. * Handle failures with goto instead of break. Update error logging. * Add --max-sessions, --session-timeout args. Enforce max sessions. * Changes to make session.go testable. Conn created with Pipe() used in testing has no unique Addr. * Add tests for headers, headers.subscribe, address.subscribe. * HashXStatus, HashXMempoolStatus not populated by default. Fix GetStatus(). * Use time.Ticker object to drive management activity.
103 lines
2.2 KiB
Go
103 lines
2.2 KiB
Go
package server
|
|
|
|
import (
|
|
"encoding/binary"
|
|
"net"
|
|
|
|
"github.com/lbryio/herald.go/internal"
|
|
"github.com/sirupsen/logrus"
|
|
)
|
|
|
|
const NotifierResponseLength = 40
|
|
|
|
// AddHeightSub adds a new height subscriber
|
|
func (s *Server) AddHeightSub(addr net.Addr, conn net.Conn) {
|
|
s.HeightSubsMut.Lock()
|
|
defer s.HeightSubsMut.Unlock()
|
|
s.HeightSubs[addr] = conn
|
|
}
|
|
|
|
// DoNotify sends a notification to all height subscribers
|
|
func (s *Server) DoNotify(heightHash *internal.HeightHash) error {
|
|
buff := make([]byte, NotifierResponseLength)
|
|
toDelete := make([]net.Addr, 0)
|
|
|
|
s.HeightSubsMut.RLock()
|
|
for addr, conn := range s.HeightSubs {
|
|
// struct.pack(b'>Q32s', height, block_hash)
|
|
binary.BigEndian.PutUint64(buff, heightHash.Height)
|
|
copy(buff[8:], heightHash.BlockHash[:32])
|
|
logrus.Tracef("notifying %s", addr)
|
|
n, err := conn.Write(buff)
|
|
if err != nil {
|
|
logrus.Warn(err)
|
|
toDelete = append(toDelete, addr)
|
|
}
|
|
if n != NotifierResponseLength {
|
|
logrus.Warn("not all bytes written")
|
|
}
|
|
}
|
|
s.HeightSubsMut.RUnlock()
|
|
|
|
if len(toDelete) > 0 {
|
|
s.HeightSubsMut.Lock()
|
|
for _, v := range toDelete {
|
|
delete(s.HeightSubs, v)
|
|
}
|
|
s.HeightSubsMut.Unlock()
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// RunNotifier Runs the notfying action forever
|
|
func (s *Server) RunNotifier() error {
|
|
for notification := range s.NotifierChan {
|
|
switch notification.(type) {
|
|
case internal.HeightHash:
|
|
heightHash, _ := notification.(internal.HeightHash)
|
|
s.DoNotify(&heightHash)
|
|
}
|
|
s.sessionManager.doNotify(notification)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// NotifierServer implements the TCP protocol for height/blockheader notifications
|
|
func (s *Server) NotifierServer() error {
|
|
address := ":" + s.Args.NotifierPort
|
|
addr, err := net.ResolveTCPAddr("tcp", address)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
listen, err := net.ListenTCP("tcp", addr)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
defer listen.Close()
|
|
|
|
for {
|
|
|
|
logrus.Info("Waiting for connection")
|
|
conn, err := listen.Accept()
|
|
if err != nil {
|
|
logrus.Warn(err)
|
|
continue
|
|
}
|
|
|
|
addr := conn.RemoteAddr()
|
|
|
|
logrus.Println(addr)
|
|
|
|
// _, err = conn.Write([]byte(addr.String()))
|
|
// if err != nil {
|
|
// logrus.Warn(err)
|
|
// continue
|
|
// }
|
|
|
|
go s.AddHeightSub(addr, conn)
|
|
}
|
|
}
|