Merge pull request #11 from github/session

rename Server to Session and simplify API

client.go

@@ -8,13 +8,10 @@ import (
 	"golang.org/x/crypto/ssh"
 )
 
-// A Client capable of joining a liveshare connection
+// A Client capable of joining a Live Share workspace.
 type Client struct {
 	connection Connection
 	tlsConfig  *tls.Config
-
-	ssh *sshSession
-	rpc *rpcClient
 }
 
 // A ClientOption is a function that modifies a client
@@ -52,31 +49,26 @@ func WithTLSConfig(tlsConfig *tls.Config) ClientOption {
 	}
 }
 
-// Join is a method that joins the client to the liveshare session
-func (c *Client) Join(ctx context.Context) (err error) {
+// JoinWorkspace connects the client to the server's Live Share
+// workspace and returns a session representing their connection.
+func (c *Client) JoinWorkspace(ctx context.Context) (*Session, error) {
 	clientSocket := newSocket(c.connection, c.tlsConfig)
 	if err := clientSocket.connect(ctx); err != nil {
-		return fmt.Errorf("error connecting websocket: %v", err)
+		return nil, fmt.Errorf("error connecting websocket: %v", err)
 	}
 
-	c.ssh = newSshSession(c.connection.SessionToken, clientSocket)
-	if err := c.ssh.connect(ctx); err != nil {
-		return fmt.Errorf("error connecting to ssh session: %v", err)
+	ssh := newSSHSession(c.connection.SessionToken, clientSocket)
+	if err := ssh.connect(ctx); err != nil {
+		return nil, fmt.Errorf("error connecting to ssh session: %v", err)
 	}
 
-	c.rpc = newRPCClient(c.ssh)
-	c.rpc.connect(ctx)
-
-	_, err = c.joinWorkspace(ctx)
-	if err != nil {
-		return fmt.Errorf("error joining Live Share workspace: %v", err)
+	rpc := newRPCClient(ssh)
+	rpc.connect(ctx)
+	if _, err := c.joinWorkspace(ctx, rpc); err != nil {
+		return nil, fmt.Errorf("error joining Live Share workspace: %v", err)
 	}
 
-	return nil
-}
-
-func (c *Client) hasJoined() bool {
-	return c.ssh != nil && c.rpc != nil
+	return &Session{ssh: ssh, rpc: rpc}, nil
 }
 
 type clientCapabilities struct {
@@ -94,7 +86,7 @@ type joinWorkspaceResult struct {
 	SessionNumber int `json:"sessionNumber"`
 }
 
-func (c *Client) joinWorkspace(ctx context.Context) (*joinWorkspaceResult, error) {
+func (c *Client) joinWorkspace(ctx context.Context, rpc *rpcClient) (*joinWorkspaceResult, error) {
 	args := joinWorkspaceArgs{
 		ID:                      c.connection.SessionID,
 		ConnectionMode:          "local",
@@ -105,21 +97,21 @@ func (c *Client) joinWorkspace(ctx context.Context) (*joinWorkspaceResult, error
 	}
 
 	var result joinWorkspaceResult
-	if err := c.rpc.do(ctx, "workspace.joinWorkspace", &args, &result); err != nil {
+	if err := rpc.do(ctx, "workspace.joinWorkspace", &args, &result); err != nil {
 		return nil, fmt.Errorf("error making workspace.joinWorkspace call: %v", err)
 	}
 
 	return &result, nil
 }
 
-func (c *Client) openStreamingChannel(ctx context.Context, streamName, condition string) (ssh.Channel, error) {
+func (s *Session) openStreamingChannel(ctx context.Context, streamName, condition string) (ssh.Channel, error) {
 	args := getStreamArgs{streamName, condition}
 	var streamID string
-	if err := c.rpc.do(ctx, "streamManager.getStream", args, &streamID); err != nil {
+	if err := s.rpc.do(ctx, "streamManager.getStream", args, &streamID); err != nil {
 		return nil, fmt.Errorf("error getting stream id: %v", err)
 	}
 
-	channel, reqs, err := c.ssh.conn.OpenChannel("session", nil)
+	channel, reqs, err := s.ssh.conn.OpenChannel("session", nil)
 	if err != nil {
 		return nil, fmt.Errorf("error opening ssh channel for transport: %v", err)
 	}

client_test.go

@@ -43,7 +43,7 @@ func TestNewClientWithInvalidConnection(t *testing.T) {
 	}
 }
 
-func TestClientJoin(t *testing.T) {
+func TestJoinSession(t *testing.T) {
 	connection := Connection{
 		SessionID:    "session-id",
 		SessionToken: "session-token",
@@ -90,10 +90,12 @@ func TestClientJoin(t *testing.T) {
 
 	done := make(chan error)
 	go func() {
-		if err := client.Join(ctx); err != nil {
-			done <- fmt.Errorf("error joining client: %v", err)
+		session, err := client.JoinWorkspace(ctx)
+		if err != nil {
+			done <- fmt.Errorf("error joining workspace: %v", err)
 			return
 		}
+		_ = session
 
 		done <- nil
 	}()

port_forwarder.go

@@ -7,20 +7,17 @@ import (
 	"net"
 )
 
-// A PortForwarder forwards TCP traffic between a port on a remote
-// LiveShare host and a local port.
+// A PortForwarder forwards TCP traffic between a local TCP port and a LiveShare session.
 type PortForwarder struct {
-	client *Client
-	server *Server
-	port   int
+	session *Session
+	port    int
 }
 
-// NewPortForwarder creates a new PortForwarder that connects a given client, server and port.
-func NewPortForwarder(client *Client, server *Server, port int) *PortForwarder {
+// NewPortForwarder creates a new PortForwarder for a given Live Share session and local TCP port.
+func NewPortForwarder(session *Session, port int) *PortForwarder {
 	return &PortForwarder{
-		client: client,
-		server: server,
-		port:   port,
+		session: session,
+		port:    port,
 	}
 }
 
@@ -87,7 +84,7 @@ func awaitError(ctx context.Context, errc <-chan error) error {
 func (l *PortForwarder) handleConnection(ctx context.Context, conn io.ReadWriteCloser) (err error) {
 	defer safeClose(conn, &err)
 
-	channel, err := l.client.openStreamingChannel(ctx, l.server.streamName, l.server.streamCondition)
+	channel, err := l.session.openStreamingChannel(ctx, l.session.streamName, l.session.streamCondition)
 	if err != nil {
 		return fmt.Errorf("error opening streaming channel for new connection: %v", err)
 	}

port_forwarder_test.go

@@ -15,16 +15,12 @@ import (
 )
 
 func TestNewPortForwarder(t *testing.T) {
-	testServer, client, err := makeMockJoinedClient()
+	testServer, session, err := makeMockSession()
 	if err != nil {
 		t.Errorf("create mock client: %v", err)
 	}
 	defer testServer.Close()
-	server, err := NewServer(client)
-	if err != nil {
-		t.Errorf("create new server: %v", err)
-	}
-	pf := NewPortForwarder(client, server, 80)
+	pf := NewPortForwarder(session, 80)
 	if pf == nil {
 		t.Error("port forwarder is nil")
 	}
@@ -40,28 +36,23 @@ func TestPortForwarderStart(t *testing.T) {
 	}
 
 	stream := bytes.NewBufferString("stream-data")
-	testServer, client, err := makeMockJoinedClient(
+	testServer, session, err := makeMockSession(
 		livesharetest.WithService("serverSharing.startSharing", serverSharing),
 		livesharetest.WithService("streamManager.getStream", getStream),
 		livesharetest.WithStream("stream-id", stream),
 	)
 	if err != nil {
-		t.Errorf("create mock client: %v", err)
+		t.Errorf("create mock session: %v", err)
 	}
 	defer testServer.Close()
 
-	server, err := NewServer(client)
-	if err != nil {
-		t.Errorf("create new server: %v", err)
-	}
-
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
-	pf := NewPortForwarder(client, server, 8000)
+	pf := NewPortForwarder(session, 8000)
 	done := make(chan error)
 
 	go func() {
-		if err := server.StartSharing(ctx, "http", 8000); err != nil {
+		if err := session.StartSharing(ctx, "http", 8000); err != nil {
 			done <- fmt.Errorf("start sharing: %v", err)
 		}
 		done <- pf.Forward(ctx)

rpc.go

@@ -21,6 +21,7 @@ func newRPCClient(conn io.ReadWriteCloser) *rpcClient {
 
 func (r *rpcClient) connect(ctx context.Context) {
 	stream := jsonrpc2.NewBufferedStream(r.conn, jsonrpc2.VSCodeObjectCodec{})
+	// TODO(adonovan): fix: ensure r.Conn is eventually Closed!
 	r.Conn = jsonrpc2.NewConn(ctx, stream, r.handler)
 }
 

server.go

@@ -1,77 +0,0 @@
-package liveshare
-
-import (
-	"context"
-	"errors"
-	"fmt"
-	"strconv"
-)
-
-// A Server represents the liveshare host and container server
-type Server struct {
-	client                      *Client
-	port                        int
-	streamName, streamCondition string
-}
-
-// NewServer creates a new Server with a given Client
-func NewServer(client *Client) (*Server, error) {
-	if !client.hasJoined() {
-		return nil, errors.New("client must join before creating server")
-	}
-
-	return &Server{client: client}, nil
-}
-
-// Port represents an open port on the container
-type Port struct {
-	SourcePort                       int    `json:"sourcePort"`
-	DestinationPort                  int    `json:"destinationPort"`
-	SessionName                      string `json:"sessionName"`
-	StreamName                       string `json:"streamName"`
-	StreamCondition                  string `json:"streamCondition"`
-	BrowseURL                        string `json:"browseUrl"`
-	IsPublic                         bool   `json:"isPublic"`
-	IsTCPServerConnectionEstablished bool   `json:"isTCPServerConnectionEstablished"`
-	HasTSLHandshakePassed            bool   `json:"hasTSLHandshakePassed"`
-}
-
-// StartSharing tells the liveshare host to start sharing the port from the container
-func (s *Server) StartSharing(ctx context.Context, protocol string, port int) error {
-	s.port = port
-
-	var response Port
-	if err := s.client.rpc.do(ctx, "serverSharing.startSharing", []interface{}{
-		port, protocol, fmt.Sprintf("http://localhost:%s", strconv.Itoa(port)),
-	}, &response); err != nil {
-		return err
-	}
-
-	s.streamName = response.StreamName
-	s.streamCondition = response.StreamCondition
-
-	return nil
-}
-
-// Ports is a slice of Port pointers
-type Ports []*Port
-
-// GetSharedServers returns a list of available/open ports from the container
-func (s *Server) GetSharedServers(ctx context.Context) (Ports, error) {
-	var response Ports
-	if err := s.client.rpc.do(ctx, "serverSharing.getSharedServers", []string{}, &response); err != nil {
-		return nil, err
-	}
-
-	return response, nil
-}
-
-// UpdateSharedVisibility controls port permissions and whether it can be accessed publicly
-// via the Browse URL
-func (s *Server) UpdateSharedVisibility(ctx context.Context, port int, public bool) error {
-	if err := s.client.rpc.do(ctx, "serverSharing.updateSharedServerVisibility", []interface{}{port, public}, nil); err != nil {
-		return err
-	}
-
-	return nil
-}

session.go

@@ -0,0 +1,69 @@
+package liveshare
+
+import (
+	"context"
+	"fmt"
+)
+
+// A Session represents the session between a connected Live Share client and server.
+type Session struct {
+	ssh *sshSession
+	rpc *rpcClient
+
+	// TODO(adonovan): fix: avoid data race of state accessed by
+	// multiple calls to StartSharing and concurrent calls to
+	// PortForwarder. Perhaps combine the two operations in the API?
+	streamName, streamCondition string
+}
+
+// Port describes a port exposed by the container.
+type Port struct {
+	SourcePort                       int    `json:"sourcePort"`
+	DestinationPort                  int    `json:"destinationPort"`
+	SessionName                      string `json:"sessionName"`
+	StreamName                       string `json:"streamName"`
+	StreamCondition                  string `json:"streamCondition"`
+	BrowseURL                        string `json:"browseUrl"`
+	IsPublic                         bool   `json:"isPublic"`
+	IsTCPServerConnectionEstablished bool   `json:"isTCPServerConnectionEstablished"`
+	HasTSLHandshakePassed            bool   `json:"hasTSLHandshakePassed"`
+	// ^^^
+	// TODO(adonovan): fix possible typo in field name, and audit others.
+}
+
+// StartSharing tells the Live Share host to start sharing the specified port from the container.
+// The sessionName describes the purpose of the port or service.
+func (s *Session) StartSharing(ctx context.Context, sessionName string, port int) error {
+	var response Port
+	if err := s.rpc.do(ctx, "serverSharing.startSharing", []interface{}{
+		port, sessionName, fmt.Sprintf("http://localhost:%d", port),
+	}, &response); err != nil {
+		return err
+	}
+
+	s.streamName = response.StreamName
+	s.streamCondition = response.StreamCondition
+
+	return nil
+}
+
+// GetSharedServers returns a description of each container port
+// shared by a prior call to StartSharing by some client.
+func (s *Session) GetSharedServers(ctx context.Context) ([]*Port, error) {
+	var response []*Port
+	if err := s.rpc.do(ctx, "serverSharing.getSharedServers", []string{}, &response); err != nil {
+		return nil, err
+	}
+
+	return response, nil
+}
+
+// UpdateSharedVisibility controls port permissions and whether it can be accessed publicly
+// via the Browse URL
+func (s *Session) UpdateSharedVisibility(ctx context.Context, port int, public bool) error {
+	if err := s.rpc.do(ctx, "serverSharing.updateSharedServerVisibility", []interface{}{port, public}, nil); err != nil {
+		return err
+	}
+
+	return nil
+}

session_test.go

@@ -13,17 +13,7 @@ import (
 	"github.com/sourcegraph/jsonrpc2"
 )
 
-func TestNewServerWithNotJoinedClient(t *testing.T) {
-	client, err := NewClient()
-	if err != nil {
-		t.Errorf("error creating new client: %v", err)
-	}
-	if _, err := NewServer(client); err == nil {
-		t.Error("expected error")
-	}
-}
-
-func makeMockJoinedClient(opts ...livesharetest.ServerOption) (*livesharetest.Server, *Client, error) {
+func makeMockSession(opts ...livesharetest.ServerOption) (*livesharetest.Server, *Session, error) {
 	connection := Connection{
 		SessionID:    "session-id",
 		SessionToken: "session-token",
@@ -47,25 +37,11 @@ func makeMockJoinedClient(opts ...livesharetest.ServerOption) (*livesharetest.Se
 		return nil, nil, fmt.Errorf("error creating new client: %v", err)
 	}
 	ctx := context.Background()
-	if err := client.Join(ctx); err != nil {
-		return nil, nil, fmt.Errorf("error joining client: %v", err)
-	}
-	return testServer, client, nil
-}
-
-func TestNewServer(t *testing.T) {
-	testServer, client, err := makeMockJoinedClient()
-	defer testServer.Close()
+	session, err := client.JoinWorkspace(ctx)
 	if err != nil {
-		t.Errorf("error creating mock joined client: %v", err)
-	}
-	server, err := NewServer(client)
-	if err != nil {
-		t.Errorf("error creating new server: %v", err)
-	}
-	if server == nil {
-		t.Error("server is nil")
+		return nil, nil, fmt.Errorf("error joining workspace: %v", err)
 	}
+	return testServer, session, nil
 }
 
 func TestServerStartSharing(t *testing.T) {
@@ -95,25 +71,21 @@ func TestServerStartSharing(t *testing.T) {
 		}
 		return Port{StreamName: "stream-name", StreamCondition: "stream-condition"}, nil
 	}
-	testServer, client, err := makeMockJoinedClient(
+	testServer, session, err := makeMockSession(
 		livesharetest.WithService("serverSharing.startSharing", startSharing),
 	)
 	defer testServer.Close()
 	if err != nil {
-		t.Errorf("error creating mock joined client: %v", err)
-	}
-	server, err := NewServer(client)
-	if err != nil {
-		t.Errorf("error creating new server: %v", err)
+		t.Errorf("error creating mock session: %v", err)
 	}
 	ctx := context.Background()
 
 	done := make(chan error)
 	go func() {
-		if err := server.StartSharing(ctx, serverProtocol, serverPort); err != nil {
+		if err := session.StartSharing(ctx, serverProtocol, serverPort); err != nil {
 			done <- fmt.Errorf("error sharing server: %v", err)
 		}
-		if server.streamName == "" || server.streamCondition == "" {
+		if session.streamName == "" || session.streamCondition == "" {
 			done <- errors.New("stream name or condition is blank")
 		}
 		done <- nil
@@ -136,23 +108,19 @@ func TestServerGetSharedServers(t *testing.T) {
 		StreamCondition: "stream-condition",
 	}
 	getSharedServers := func(req *jsonrpc2.Request) (interface{}, error) {
-		return Ports{&sharedServer}, nil
+		return []*Port{&sharedServer}, nil
 	}
-	testServer, client, err := makeMockJoinedClient(
+	testServer, session, err := makeMockSession(
 		livesharetest.WithService("serverSharing.getSharedServers", getSharedServers),
 	)
 	if err != nil {
-		t.Errorf("error creating new mock client: %v", err)
+		t.Errorf("error creating mock session: %v", err)
 	}
 	defer testServer.Close()
-	server, err := NewServer(client)
-	if err != nil {
-		t.Errorf("error creating new server: %v", err)
-	}
 	ctx := context.Background()
 	done := make(chan error)
 	go func() {
-		ports, err := server.GetSharedServers(ctx)
+		ports, err := session.GetSharedServers(ctx)
 		if err != nil {
 			done <- fmt.Errorf("error getting shared servers: %v", err)
 		}
@@ -206,25 +174,17 @@ func TestServerUpdateSharedVisibility(t *testing.T) {
 		}
 		return nil, nil
 	}
-	testServer, client, err := makeMockJoinedClient(
+	testServer, session, err := makeMockSession(
 		livesharetest.WithService("serverSharing.updateSharedServerVisibility", updateSharedVisibility),
 	)
 	if err != nil {
-		t.Errorf("creating new mock client: %v", err)
+		t.Errorf("creating mock session: %v", err)
 	}
 	defer testServer.Close()
-	server, err := NewServer(client)
-	if err != nil {
-		t.Errorf("creating server: %v", err)
-	}
 	ctx := context.Background()
 	done := make(chan error)
 	go func() {
-		if err := server.UpdateSharedVisibility(ctx, 80, true); err != nil {
-			done <- err
-			return
-		}
-		done <- nil
+		done <- session.UpdateSharedVisibility(ctx, 80, true)
 	}()
 	select {
 	case err := <-testServer.Err():

ssh.go

@@ -19,7 +19,7 @@ type sshSession struct {
 	writer io.Writer
 }
 
-func newSshSession(token string, socket net.Conn) *sshSession {
+func newSSHSession(token string, socket net.Conn) *sshSession {
 	return &sshSession{token: token, socket: socket}
 }
 

ssh_server.go

@@ -2,18 +2,14 @@ package liveshare
 
 import (
 	"context"
-	"errors"
 )
 
 type SSHServer struct {
-	client *Client
+	session *Session
 }
 
-func NewSSHServer(client *Client) (*SSHServer, error) {
-	if !client.hasJoined() {
-		return nil, errors.New("client must join before creating server")
-	}
-	return &SSHServer{client: client}, nil
+func (session *Session) SSHServer() *SSHServer {
+	return &SSHServer{session: session}
 }
 
 type SSHServerStartResult struct {
@@ -23,12 +19,12 @@ type SSHServerStartResult struct {
 	Message    string `json:"message"`
 }
 
-func (s *SSHServer) StartRemoteServer(ctx context.Context) (SSHServerStartResult, error) {
+func (s *SSHServer) StartRemoteServer(ctx context.Context) (*SSHServerStartResult, error) {
 	var response SSHServerStartResult
 
-	if err := s.client.rpc.do(ctx, "ISshServerHostService.startRemoteServer", []string{}, &response); err != nil {
-		return response, err
+	if err := s.session.rpc.do(ctx, "ISshServerHostService.startRemoteServer", []string{}, &response); err != nil {
+		return nil, err
 	}
 
-	return response, nil
+	return &response, nil
 }

terminal.go

@@ -2,7 +2,6 @@ package liveshare
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"io"
 
@@ -11,17 +10,11 @@ import (
 )
 
 type Terminal struct {
-	client *Client
+	session *Session
 }
 
-func NewTerminal(client *Client) (*Terminal, error) {
-	if !client.hasJoined() {
-		return nil, errors.New("client must join before creating terminal")
-	}
-
-	return &Terminal{
-		client: client,
-	}, nil
+func NewTerminal(session *Session) *Terminal {
+	return &Terminal{session: session}
 }
 
 type TerminalCommand struct {
@@ -73,16 +66,16 @@ func (t TerminalCommand) Run(ctx context.Context) (io.ReadCloser, error) {
 	}
 
 	started := make(chan struct{})
-	t.terminal.client.rpc.handler.registerEventHandler("terminal.terminalStarted", func(*jsonrpc2.Request) {
+	t.terminal.session.rpc.handler.registerEventHandler("terminal.terminalStarted", func(*jsonrpc2.Request) {
 		close(started)
 	})
 	var result startTerminalResult
-	if err := t.terminal.client.rpc.do(ctx, "terminal.startTerminal", &args, &result); err != nil {
+	if err := t.terminal.session.rpc.do(ctx, "terminal.startTerminal", &args, &result); err != nil {
 		return nil, fmt.Errorf("error making terminal.startTerminal call: %v", err)
 	}
 	<-started
 
-	channel, err := t.terminal.client.openStreamingChannel(ctx, result.StreamName, result.StreamCondition)
+	channel, err := t.terminal.session.openStreamingChannel(ctx, result.StreamName, result.StreamCondition)
 	if err != nil {
 		return nil, fmt.Errorf("error opening streaming channel: %v", err)
 	}
@@ -106,10 +99,10 @@ func (t terminalReadCloser) Read(b []byte) (int, error) {
 
 func (t terminalReadCloser) Close() error {
 	stopped := make(chan struct{})
-	t.terminalCommand.terminal.client.rpc.handler.registerEventHandler("terminal.terminalStopped", func(*jsonrpc2.Request) {
+	t.terminalCommand.terminal.session.rpc.handler.registerEventHandler("terminal.terminalStopped", func(*jsonrpc2.Request) {
 		close(stopped)
 	})
-	if err := t.terminalCommand.terminal.client.rpc.do(context.Background(), "terminal.stopTerminal", []int{t.terminalID}, nil); err != nil {
+	if err := t.terminalCommand.terminal.session.rpc.do(context.Background(), "terminal.stopTerminal", []int{t.terminalID}, nil); err != nil {
 		return fmt.Errorf("error making terminal.stopTerminal call: %v", err)
 	}