akovalenko/lnd.xprv
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Merge pull request #2140 from cfromknecht/wtserver-move

Browse Source 
[watchtower/wtserver] rename server package, add godocs, general code health
This commit is contained in:
Olaoluwa Osuntokun 2018-11-28 15:07:01 -08:00 committed by GitHub
commit fd82200a15
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 65 additions and 51 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
watchtower/server/interface.go → watchtower/wtserver/interface.go
View File

@ -1,4 +1,4 @@
package server
package wtserver
import (
"io"

4
watchtower/server/log.go → watchtower/wtserver/log.go
View File

@ -1,4 +1,4 @@
package server
package wtserver
import (
"github.com/btcsuite/btclog"
@ -12,7 +12,7 @@ var log btclog.Logger
// The default amount of logging is none.
func init() {
UseLogger(build.NewSubLogger("WTSV", nil))
UseLogger(build.NewSubLogger("WTWR", nil))
}
// DisableLog disables all library log output. Logging output is disabled

2
watchtower/server/mock.go → watchtower/wtserver/mock.go
View File

@ -1,6 +1,6 @@
// +build dev
package server
package wtserver
import (
"fmt"

70
watchtower/server/server.go → watchtower/wtserver/server.go
View File

@ -1,4 +1,4 @@
package server
package wtserver
import (
"bytes"
@ -108,8 +108,12 @@ func (s *Server) Start() error {
return nil
}
log.Infof("Starting watchtower server")
s.connMgr.Start()
log.Infof("Watchtower server started successfully")
return nil
}
@ -120,11 +124,15 @@ func (s *Server) Stop() error {
return nil
}
log.Infof("Stopping watchtower server")
s.connMgr.Stop()
close(s.quit)
s.wg.Wait()
log.Infof("Watchtower server stopped successfully")
return nil
}
@ -287,6 +295,8 @@ func (s *Server) handleClient(peer Peer) {
}
}
// handleInit accepts the local and remote Init messages, and verifies that the
// client is not requesting any required features that are unknown to the tower.
func (s *Server) handleInit(localInit, remoteInit *wtwire.Init) error {
remoteLocalFeatures := lnwire.NewFeatureVector(
remoteInit.LocalFeatures, wtwire.LocalFeatures,
@ -312,33 +322,6 @@ func (s *Server) handleInit(localInit, remoteInit *wtwire.Init) error {
return nil
}
func (s *Server) readMessage(peer Peer) (wtwire.Message, error) {
// Set a read timeout to ensure we drop the client if not sent in a
// timely manner.
err := peer.SetReadDeadline(time.Now().Add(s.cfg.ReadTimeout))
if err != nil {
err = fmt.Errorf("unable to set read deadline: %v", err)
return nil, err
}
// Pull the next message off the wire, and parse it according to the
// watchtower wire specification.
rawMsg, err := peer.ReadNextMessage()
if err != nil {
err = fmt.Errorf("unable to read message: %v", err)
return nil, err
}
msgReader := bytes.NewReader(rawMsg)
nextMsg, err := wtwire.ReadMessage(msgReader, 0)
if err != nil {
err = fmt.Errorf("unable to parse message: %v", err)
return nil, err
}
return nextMsg, nil
}
// handleCreateSession processes a CreateSession message from the peer, and returns
// a CreateSessionReply in response. This method will only succeed if no existing
// session info is known about the session id. If an existing session is found,
@ -550,6 +533,37 @@ func (s *Server) replyStateUpdate(peer Peer, id *wtdb.SessionID,
}
}
// readMessage receives and parses the next message from the given Peer. An
// error is returned if a message is not received before the server's read
// timeout, the read off the wire failed, or the message could not be
// deserialized.
func (s *Server) readMessage(peer Peer) (wtwire.Message, error) {
// Set a read timeout to ensure we drop the client if not sent in a
// timely manner.
err := peer.SetReadDeadline(time.Now().Add(s.cfg.ReadTimeout))
if err != nil {
err = fmt.Errorf("unable to set read deadline: %v", err)
return nil, err
}
// Pull the next message off the wire, and parse it according to the
// watchtower wire specification.
rawMsg, err := peer.ReadNextMessage()
if err != nil {
err = fmt.Errorf("unable to read message: %v", err)
return nil, err
}
msgReader := bytes.NewReader(rawMsg)
nextMsg, err := wtwire.ReadMessage(msgReader, 0)
if err != nil {
err = fmt.Errorf("unable to parse message: %v", err)
return nil, err
}
return nextMsg, nil
}
// sendMessage sends a watchtower wire message to the target peer.
func (s *Server) sendMessage(peer Peer, msg wtwire.Message) error {
// TODO(conner): use buffer pool?

38
watchtower/server/server_test.go → watchtower/wtserver/server_test.go
View File

@ -1,6 +1,6 @@
// +build dev
package server_test
package wtserver_test
import (
"bytes"
@ -12,8 +12,8 @@ import (
"github.com/btcsuite/btcd/chaincfg"
"github.com/btcsuite/btcutil"
"github.com/lightningnetwork/lnd/lnwire"
"github.com/lightningnetwork/lnd/watchtower/server"
"github.com/lightningnetwork/lnd/watchtower/wtdb"
"github.com/lightningnetwork/lnd/watchtower/wtserver"
"github.com/lightningnetwork/lnd/watchtower/wtwire"
)
@ -36,8 +36,8 @@ func randPubKey(t *testing.T) *btcec.PublicKey {
// initServer creates and starts a new server using the server.DB and timeout.
// If the provided database is nil, a mock db will be used.
func initServer(t *testing.T, db server.DB,
timeout time.Duration) server.Interface {
func initServer(t *testing.T, db wtserver.DB,
timeout time.Duration) wtserver.Interface {
t.Helper()
@ -45,7 +45,7 @@ func initServer(t *testing.T, db server.DB,
db = wtdb.NewMockDB()
}
s, err := server.New(&server.Config{
s, err := wtserver.New(&wtserver.Config{
DB: db,
ReadTimeout: timeout,
WriteTimeout: timeout,
@ -79,8 +79,8 @@ func TestServerOnlyAcceptOnePeer(t *testing.T) {
// Create two peers using the same session id.
peerPub := randPubKey(t)
peer1 := server.NewMockPeer(peerPub, nil, 0)
peer2 := server.NewMockPeer(peerPub, nil, 0)
peer1 := wtserver.NewMockPeer(peerPub, nil, 0)
peer2 := wtserver.NewMockPeer(peerPub, nil, 0)
// Serialize a Init message to be sent by both peers.
init := wtwire.NewInitMessage(
@ -106,8 +106,8 @@ func TestServerOnlyAcceptOnePeer(t *testing.T) {
// Try to send a message on either peer, and record the opposite peer as
// the one we assume to be rejected.
var (
rejectedPeer *server.MockPeer
acceptedPeer *server.MockPeer
rejectedPeer *wtserver.MockPeer
acceptedPeer *wtserver.MockPeer
)
select {
case peer1.IncomingMsgs <- msg:
@ -192,7 +192,7 @@ func testServerCreateSession(t *testing.T, i int, test createSessionTestCase) {
// Create a new client and connect to server.
peerPub := randPubKey(t)
peer := server.NewMockPeer(peerPub, nil, 0)
peer := wtserver.NewMockPeer(peerPub, nil, 0)
connect(t, i, s, peer, test.initMsg, timeoutDuration)
// Send the CreateSession message, and wait for a reply.
@ -220,7 +220,7 @@ func testServerCreateSession(t *testing.T, i int, test createSessionTestCase) {
// Simulate a peer with the same session id connection to the server
// again.
peer = server.NewMockPeer(peerPub, nil, 0)
peer = wtserver.NewMockPeer(peerPub, nil, 0)
connect(t, i, s, peer, test.initMsg, timeoutDuration)
// Send the _same_ CreateSession message as the first attempt.
@ -492,7 +492,7 @@ func testServerStateUpdates(t *testing.T, i int, test stateUpdateTestCase) {
// Create a new client and connect to the server.
peerPub := randPubKey(t)
peer := server.NewMockPeer(peerPub, nil, 0)
peer := wtserver.NewMockPeer(peerPub, nil, 0)
connect(t, i, s, peer, test.initMsg, timeoutDuration)
// Register a session for this client to use in the subsequent tests.
@ -512,7 +512,7 @@ func testServerStateUpdates(t *testing.T, i int, test stateUpdateTestCase) {
// Now that the original connection has been closed, connect a new
// client with the same session id.
peer = server.NewMockPeer(peerPub, nil, 0)
peer = wtserver.NewMockPeer(peerPub, nil, 0)
connect(t, i, s, peer, test.initMsg, timeoutDuration)
// Send the intended StateUpdate messages in series.
@ -523,7 +523,7 @@ func testServerStateUpdates(t *testing.T, i int, test stateUpdateTestCase) {
if update == nil {
assertConnClosed(t, peer, 2*timeoutDuration)
peer = server.NewMockPeer(peerPub, nil, 0)
peer = wtserver.NewMockPeer(peerPub, nil, 0)
connect(t, i, s, peer, test.initMsg, timeoutDuration)
continue
@ -547,7 +547,7 @@ func testServerStateUpdates(t *testing.T, i int, test stateUpdateTestCase) {
assertConnClosed(t, peer, 2*timeoutDuration)
}
func connect(t *testing.T, i int, s server.Interface, peer *server.MockPeer,
func connect(t *testing.T, i int, s wtserver.Interface, peer *wtserver.MockPeer,
initMsg *wtwire.Init, timeout time.Duration) {
s.InboundPeerConnected(peer)
@ -555,9 +555,9 @@ func connect(t *testing.T, i int, s server.Interface, peer *server.MockPeer,
recvReply(t, i, "Init", peer, timeout)
}
// sendMsg sends a wtwire.Message message via a server.MockPeer.
// sendMsg sends a wtwire.Message message via a wtserver.MockPeer.
func sendMsg(t *testing.T, i int, msg wtwire.Message,
peer *server.MockPeer, timeout time.Duration) {
peer *wtserver.MockPeer, timeout time.Duration) {
t.Helper()
@ -579,7 +579,7 @@ func sendMsg(t *testing.T, i int, msg wtwire.Message,
// expected reply type. The supported replies are CreateSessionReply and
// StateUpdateReply.
func recvReply(t *testing.T, i int, name string,
peer *server.MockPeer, timeout time.Duration) wtwire.Message {
peer *wtserver.MockPeer, timeout time.Duration) wtwire.Message {
t.Helper()
@ -623,7 +623,7 @@ func recvReply(t *testing.T, i int, name string,
// assertConnClosed checks that the peer's connection is closed before the
// timeout expires.
func assertConnClosed(t *testing.T, peer *server.MockPeer, duration time.Duration) {
func assertConnClosed(t *testing.T, peer *wtserver.MockPeer, duration time.Duration) {
t.Helper()
select {