client: register endpoint listener

client/client.go

@@ -2,9 +2,12 @@ package piko
 
 import (
 	"context"
+	"encoding/json"
+	"fmt"
 	"net/url"
 	"time"
 
+	"github.com/andydunstall/piko/pkg/protocol"
 	"github.com/andydunstall/piko/pkg/websocket"
 	"golang.ngrok.com/muxado/v2"
 )
@@ -56,17 +59,41 @@ func Connect(ctx context.Context, opts ...ConnectOption) (*Piko, error) {
 // which accepts incoming connections for that endpoint.
 //
 // [Listener] is a [net.Listener].
-// nolint
-func (p *Piko) Listen(ctx context.Context, endpointID string) (Listener, error) {
-	p.sess.Wait()
-	return nil, nil
+func (p *Piko) Listen(_ context.Context, endpointID string) (Listener, error) {
+	req := &protocol.ListenRequest{
+		EndpointID: endpointID,
+	}
+	var resp protocol.ListenResponse
+	if err := p.rpc(protocol.RPCTypeListen, req, &resp); err != nil {
+		return nil, fmt.Errorf("rpc: %w", err)
+	}
+
+	return &listener{}, nil
 }
 
 // Close closes the connection to Piko and any open listeners.
 func (p *Piko) Close() error {
 	return nil
 }
 
+func (p *Piko) rpc(rpcType protocol.RPCType, req any, resp any) error {
+	stream, err := p.sess.OpenTypedStream(muxado.StreamType(rpcType))
+	if err != nil {
+		return fmt.Errorf("open stream: %w", err)
+	}
+	defer stream.Close()
+
+	if err := json.NewEncoder(stream).Encode(req); err != nil {
+		return fmt.Errorf("encode req: %w", err)
+	}
+
+	if err := json.NewDecoder(stream).Decode(&resp); err != nil {
+		return fmt.Errorf("decode resp: %w", err)
+	}
+
+	return nil
+}
+
 func serverURL(urlStr string) string {
 	// Already verified URL in Config.Validate.
 	u, _ := url.Parse(urlStr)

client/listener.go

@@ -13,3 +13,24 @@ type Listener interface {
 	// connections on.
 	EndpointID() string
 }
+
+type listener struct {
+}
+
+func (l *listener) Accept() (net.Conn, error) {
+	return nil, nil
+}
+
+func (l *listener) Close() error {
+	return nil
+}
+
+func (l *listener) Addr() net.Addr {
+	return nil
+}
+
+func (l *listener) EndpointID() string {
+	return ""
+}
+
+var _ Listener = &listener{}

pkg/protocol/message.go

@@ -0,0 +1,17 @@
+package protocol
+
+import "golang.ngrok.com/muxado/v2"
+
+type RPCType muxado.StreamType
+
+const (
+	RPCTypeListen RPCType = iota + 1
+)
+
+type ListenRequest struct {
+	EndpointID string `json:"endpoint_id"`
+}
+
+type ListenResponse struct {
+	EndpointID string `json:"endpoint_id"`
+}

serverv2/server/upstream/server.go

@@ -3,12 +3,14 @@ package server
 import (
 	"context"
 	"crypto/tls"
+	"encoding/json"
 	"fmt"
 	"net"
 	"net/http"
 	"time"
 
 	"github.com/andydunstall/piko/pkg/log"
+	"github.com/andydunstall/piko/pkg/protocol"
 	pikowebsocket "github.com/andydunstall/piko/pkg/websocket"
 	"github.com/gin-gonic/gin"
 	"github.com/gorilla/websocket"
@@ -107,10 +109,16 @@ func (s *Server) wsRoute(c *gin.Context) {
 	)
 
 	for {
-		_, err := heartbeat.AcceptTypedStream()
+		stream, err := heartbeat.AcceptTypedStream()
 		if err != nil {
+			s.logger.Warn("accept stream", zap.Error(err))
 			return
 		}
+		go func() {
+			if err := s.handleStream(stream); err != nil {
+				s.logger.Warn("handle stream", zap.Error(err))
+			}
+		}()
 	}
 }
 
@@ -123,6 +131,39 @@ func (s *Server) panicRoute(c *gin.Context, err any) {
 	c.AbortWithStatus(http.StatusInternalServerError)
 }
 
+func (s *Server) handleStream(stream muxado.TypedStream) error {
+	switch protocol.RPCType(stream.StreamType()) {
+	case protocol.RPCTypeListen:
+		if err := s.handleListenRequest(stream); err != nil {
+			return fmt.Errorf("listen request: %w", err)
+		}
+		return nil
+	default:
+		return fmt.Errorf("unsupported rpc type: %d", stream.StreamType())
+	}
+}
+
+func (s *Server) handleListenRequest(stream muxado.TypedStream) error {
+	var req protocol.ListenRequest
+	if err := json.NewDecoder(stream).Decode(&req); err != nil {
+		return fmt.Errorf("decode request: %w", err)
+	}
+
+	resp := protocol.ListenResponse(req)
+	if err := json.NewEncoder(stream).Encode(resp); err != nil {
+		return fmt.Errorf("decode response: %w", err)
+	}
+
+	s.logger.Info(
+		"registered listener",
+		zap.String("endpoint-id", req.EndpointID),
+	)
+
+	// TODO(andydunstall): Register.
+
+	return nil
+}
+
 func init() {
 	// Disable Gin debug logs.
 	gin.SetMode(gin.ReleaseMode)