multi: grpc ClientConnTransport impl

In this commit, a new grpc implementation of the ClientConnTransport
interface is added. Various changes are made to Client and ClientConn so
that a caller can choose which implementation of ClientConnTransport
should be used.

Author: ellemouton

cmd/wasm-client/lnd_conn.go

@@ -30,7 +30,9 @@ func mailboxRPCConnection(mailboxServer, pairingPhrase string,
 	)
 
 	ctx := context.Background()
-	transportConn, err := mailbox.NewClient(ctx, connData)
+	transportConn, err := mailbox.NewWebsocketsClient(
+		ctx, mailboxServer, connData,
+	)
 	if err != nil {
 		return nil, nil, err
 	}

itest/client_harness.go

@@ -62,7 +62,9 @@ func (c *clientHarness) start() error {
 		},
 	)
 
-	transportConn, err := mailbox.NewClient(ctx, connData)
+	transportConn, err := mailbox.NewWebsocketsClient(
+		ctx, c.serverAddr, connData,
+	)
 	if err != nil {
 		return err
 	}

mailbox/client.go

@@ -3,16 +3,23 @@ package mailbox
 import (
 	"bytes"
 	"context"
+	"fmt"
 	"net"
 	"sync"
+
+	"github.com/lightninglabs/lightning-node-connect/hashmailrpc"
+	"google.golang.org/grpc"
 )
 
 // Client manages the mailboxConn it holds and refreshes it on connection
 // retries.
 type Client struct {
+	serverHost string
+	connData   *ConnData
+
 	mailboxConn *ClientConn
 
-	connData *ConnData
+	grpcClient hashmailrpc.HashMailClient
 
 	status   ClientStatus
 	statusMu sync.Mutex
@@ -22,27 +29,56 @@ type Client struct {
 	ctx context.Context
 }
 
-// NewClient creates a new Client object which will handle the mailbox
-// connection.
-func NewClient(ctx context.Context, connData *ConnData) (*Client, error) {
+// NewGrpcClient creates a new Client object which will handle the mailbox
+// connection and will use grpc streams to connect to the mailbox.
+func NewGrpcClient(ctx context.Context, serverHost string, connData *ConnData,
+	dialOpts ...grpc.DialOption) (*Client, error) {
+
+	mailboxGrpcConn, err := grpc.Dial(serverHost, dialOpts...)
+	if err != nil {
+		return nil, fmt.Errorf("unable to connect to RPC server: %v",
+			err)
+	}
+
+	sid, err := connData.SID()
+	if err != nil {
+		return nil, err
+	}
+
+	return &Client{
+		ctx:        ctx,
+		serverHost: serverHost,
+		connData:   connData,
+		grpcClient: hashmailrpc.NewHashMailClient(mailboxGrpcConn),
+		status:     ClientStatusNotConnected,
+		sid:        sid,
+	}, nil
+}
+
+// NewWebsocketsClient creates a new Client object which will handle the mailbox
+// connection and will use websockets to connect to the mailbox.
+func NewWebsocketsClient(ctx context.Context, serverHost string,
+	connData *ConnData) (*Client, error) {
+
 	sid, err := connData.SID()
 	if err != nil {
 		return nil, err
 	}
 
 	return &Client{
-		ctx:      ctx,
-		connData: connData,
-		status:   ClientStatusNotConnected,
-		sid:      sid,
+		ctx:        ctx,
+		serverHost: serverHost,
+		connData:   connData,
+		status:     ClientStatusNotConnected,
+		sid:        sid,
 	}, nil
 }
 
 // Dial returns a net.Conn abstraction over the mailbox connection. Dial is
 // called everytime grpc retries the connection. If this is the first
 // connection, a new ClientConn will be created. Otherwise, the existing
 // connection will just be refreshed.
-func (c *Client) Dial(_ context.Context, serverHost string) (net.Conn, error) {
+func (c *Client) Dial(_ context.Context, _ string) (net.Conn, error) {
 	// If there is currently an active connection, block here until the
 	// previous connection as been closed.
 	if c.mailboxConn != nil {
@@ -73,7 +109,8 @@ func (c *Client) Dial(_ context.Context, serverHost string) (net.Conn, error) {
 
 	if c.mailboxConn == nil {
 		mailboxConn, err := NewClientConn(
-			c.ctx, c.sid, serverHost, func(status ClientStatus) {
+			c.ctx, c.sid, c.serverHost, c.grpcClient,
+			func(status ClientStatus) {
 				c.statusMu.Lock()
 				c.status = status
 				c.statusMu.Unlock()

mailbox/client_conn.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
 	"github.com/lightninglabs/lightning-node-connect/gbn"
+	"github.com/lightninglabs/lightning-node-connect/hashmailrpc"
 	"google.golang.org/protobuf/encoding/protojson"
 )
 
@@ -129,6 +130,7 @@ type ClientConn struct {
 // session identifiers. The context given as the first parameter will be used
 // throughout the connection lifetime.
 func NewClientConn(ctx context.Context, sid [64]byte, serverHost string,
+	client hashmailrpc.HashMailClient,
 	onNewStatus func(status ClientStatus)) (*ClientConn, error) {
 
 	receiveSID := GetSID(sid, true)
@@ -143,8 +145,16 @@ func NewClientConn(ctx context.Context, sid [64]byte, serverHost string,
 		receiveSID[:], sendSID[:])
 
 	ctxc, cancel := context.WithCancel(ctx)
+
+	var transport ClientConnTransport
+	if client != nil {
+		transport = newGrpcTransport(mailBoxInfo, client)
+	} else {
+		transport = newWebsocketTransport(mailBoxInfo)
+	}
+
 	c := &ClientConn{
-		transport: newWebsocketTransport(mailBoxInfo),
+		transport: transport,
 		gbnOptions: []gbn.Option{
 			gbn.WithTimeout(gbnTimeout),
 			gbn.WithHandshakeTimeout(gbnHandshakeTimeout),

mailbox/client_transport.go

@@ -229,3 +229,135 @@ func (wt *websocketTransport) CloseSend() error {
 
 	return wt.sendSocket.Close(websocket.StatusNormalClosure, "bye")
 }
+
+// grpcTransport is an implementation of ClientConnTransport that uses grpc
+// streams to connect the the mailbox.
+type grpcTransport struct {
+	*mailboxInfo
+
+	client        hashmailrpc.HashMailClient
+	receiveStream hashmailrpc.HashMail_RecvStreamClient
+	sendStream    hashmailrpc.HashMail_SendStreamClient
+}
+
+// newGrpcTransport constructs a new grpcTransport instance.
+func newGrpcTransport(mbInfo *mailboxInfo,
+	client hashmailrpc.HashMailClient) *grpcTransport {
+
+	return &grpcTransport{
+		client:      client,
+		mailboxInfo: mbInfo,
+	}
+}
+
+// Refresh creates a new ClientConnTransport with no initialised send or receive
+// connections.
+//
+// NOTE: this is part of the ClientConnTransport interface.
+func (gt *grpcTransport) Refresh() ClientConnTransport {
+	return &grpcTransport{
+		client:      gt.client,
+		mailboxInfo: gt.mailboxInfo,
+	}
+}
+
+// ReceiveConnected returns true if the transport is connected to the
+// hashmail-server receive stream.
+//
+// NOTE: this is part of the ClientConnTransport interface.
+func (gt *grpcTransport) ReceiveConnected() bool {
+	return gt.receiveStream != nil
+}
+
+// SendConnected returns true if the transport is connected to the
+// hashmail-server send stream.
+//
+// NOTE: this is part of the ClientConnTransport interface.
+func (gt *grpcTransport) SendConnected() bool {
+	return gt.sendStream != nil
+}
+
+// ConnectSend can be called in order to initialise the send-stream with
+// the hashmail-server.
+//
+// NOTE: this is part of the ClientConnTransport interface.
+func (gt *grpcTransport) ConnectSend(ctx context.Context) error {
+	sendStream, err := gt.client.SendStream(ctx)
+	if err != nil {
+		return err
+	}
+
+	gt.sendStream = sendStream
+	return nil
+}
+
+// ConnectReceive can be called in order to initialise the receive-stream with
+// the hashmail-server.
+//
+// NOTE: this is part of the ClientConnTransport interface.
+func (gt *grpcTransport) ConnectReceive(ctx context.Context) error {
+	receiveInit := &hashmailrpc.CipherBoxDesc{StreamId: gt.recvSID}
+	readStream, err := gt.client.RecvStream(ctx, receiveInit)
+	if err != nil {
+		return err
+	}
+
+	gt.receiveStream = readStream
+	return nil
+}
+
+// Recv will attempt to read data off of the underlying transport's
+// receive stream.
+//
+// NOTE: this is part of the ClientConnTransport interface.
+func (gt *grpcTransport) Recv(_ context.Context) ([]byte, bool, ClientStatus,
+	error) {
+
+	controlMsg, err := gt.receiveStream.Recv()
+	if err != nil {
+		return nil, true, statusFromError(err), err
+	}
+
+	return controlMsg.Msg, false, ClientStatusConnected, nil
+}
+
+// Send will attempt to send data on the underlying transport's send-stream.
+//
+// NOTE: this is part of the ClientConnTransport interface.
+func (gt *grpcTransport) Send(_ context.Context, streamID, payload []byte) (
+	bool, ClientStatus, error) {
+
+	err := gt.sendStream.Send(&hashmailrpc.CipherBox{
+		Desc: &hashmailrpc.CipherBoxDesc{
+			StreamId: streamID,
+		},
+		Msg: payload,
+	})
+	if err != nil {
+		return true, ClientStatusNotConnected, err
+	}
+
+	return false, ClientStatusConnected, nil
+}
+
+// CloseReceive will close the transport's connection to the receive-stream.
+//
+// NOTE: this is part of the ClientConnTransport interface.
+func (gt *grpcTransport) CloseReceive() error {
+	if gt.receiveStream == nil {
+		return nil
+	}
+
+	return gt.receiveStream.CloseSend()
+}
+
+// CloseSend will close the transport's connection to the send-stream.
+//
+// NOTE: this is part of the ClientConnTransport interface.
+func (gt *grpcTransport) CloseSend() error {
+	if gt.sendStream == nil {
+		return nil
+	}
+
+	return gt.sendStream.CloseSend()
+}