wip

Author: nicksanford

components/camera/client.go

@@ -8,7 +8,6 @@ import (
 	"image"
 	"io"
 	"os"
-	"runtime/debug"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -97,7 +96,6 @@ func NewClientFromConn(
 		associatedSubs: map[int][]rtppassthrough.SubscriptionID{},
 		logger:         logger,
 	}
-	debug.PrintStack()
 	logger.Infof("%p cameraClient %#v", c, c)
 	return cc, nil
 }
@@ -484,7 +482,11 @@ func (c *client) SubscribeRTP(
 		// PeerConnection renegotiation to add this camera's video track and have the `OnTrack`
 		// callback invoked.
 		c.logger.Infof("%p SubscribeRTP calling AddStream on %s", c, c.trackName())
-		if _, err := c.streamClient.AddStream(ctx, &streampb.AddStreamRequest{Name: c.trackName()}); err != nil {
+
+		if _, err := c.streamClient.AddStream(
+			ctx,
+			&streampb.AddStreamRequest{Name: c.trackName()},
+		); err != nil {
 			c.logger.CDebugf(ctx, "%p SubscribeRTP AddStream hit error subID: %s, trackName: %s, err: %s", c, sub.ID.String(), c.trackName(), err.Error())
 			return rtppassthrough.NilSubscription, err
 		}
@@ -643,7 +645,11 @@ func (c *client) Unsubscribe(ctx context.Context, id rtppassthrough.Subscription
 	request := &streampb.RemoveStreamRequest{Name: c.trackName()}
 	// We assume the server responds with a success if the requested `Name` is unknown/already
 	// removed.
-	if _, err := c.streamClient.RemoveStream(ctx, request); err != nil {
+
+	if _, err := c.streamClient.RemoveStream(
+		ctx,
+		request,
+	); err != nil {
 		c.logger.CWarnw(ctx, "Unsubscribe RemoveStream returned err", "trackName",
 			c.trackName(), "subID", id.String(), "err", err)
 		c.rtpPassthroughMu.Unlock()

grpc/shared_conn.go

@@ -95,22 +95,22 @@ func NewSharedConn(grpcConn rpc.ClientConn, peerConn *webrtc.PeerConnection, log
 		// We were passed in a ready connection. Only create this for when `Close` is called.
 		peerConnFailed:       make(chan struct{}),
 		onTrackCBByTrackName: make(map[string]OnTrackCB),
+		logger:               logger,
 	}
-	ret.ResetConn(&ret.grpcConn, logger)
-	// peerConn.OnTrack(func(trackRemote *webrtc.TrackRemote, rtpReceiver *webrtc.RTPReceiver) {
-	// 	ret.logger.Warnf("OnTrack called on pc: %p, %#v, trackRemote: %p, %#v, rtpReceiver: %p, %#v", peerConn, peerConn, trackRemote, trackRemote, rtpReceiver, rtpReceiver)
-	// 	ret.onTrackCBByTrackNameMu.Lock()
-	// 	onTrackCB, ok := ret.onTrackCBByTrackName[trackRemote.StreamID()]
-	// 	ret.onTrackCBByTrackNameMu.Unlock()
-	// 	if !ok {
-	// 		msg := "Callback not found for StreamID: %s, keys(resOnTrackCBs): %#v"
-	// 		ret.logger.Errorf(msg, trackRemote.StreamID(), maps.Keys(ret.onTrackCBByTrackName))
-	// 		return
-	// 	}
-	// 	onTrackCB(trackRemote, rtpReceiver)
-	// })
-
-	// debug.PrintStack()
+	ret.grpcConn.ReplaceConn(grpcConn)
+	ret.logger.Infof("OnTrack installed on %p", peerConn)
+	ret.peerConn.OnTrack(func(trackRemote *webrtc.TrackRemote, rtpReceiver *webrtc.RTPReceiver) {
+		ret.logger.Info("OnTrack called:", trackRemote.ID())
+		ret.onTrackCBByTrackNameMu.Lock()
+		onTrackCB, ok := ret.onTrackCBByTrackName[trackRemote.StreamID()]
+		ret.onTrackCBByTrackNameMu.Unlock()
+		if !ok {
+			msg := "Callback not found for StreamID: %s, keys(resOnTrackCBs): %#v"
+			ret.logger.Errorf(msg, trackRemote.StreamID(), maps.Keys(ret.onTrackCBByTrackName))
+			return
+		}
+		onTrackCB(trackRemote, rtpReceiver)
+	})
 
 	return ret
 }
@@ -184,6 +184,7 @@ func (sc *SharedConn) PeerConn() *webrtc.PeerConnection {
 // happens. But subequent calls can be entirely asynchronous to components/services accessing
 // `SharedConn` for connection objects.
 func (sc *SharedConn) ResetConn(conn rpc.ClientConn, moduleLogger logging.Logger) {
+	moduleLogger.Infof("ResetConn called on %p, sharedConn: %p", conn, sc)
 	sc.grpcConn.ReplaceConn(conn)
 	if sc.logger == nil {
 		// The first call to `ResetConn` happens before anything can access `sc.logger`. So long as
@@ -226,6 +227,7 @@ func (sc *SharedConn) ResetConn(conn rpc.ClientConn, moduleLogger logging.Logger
 		return
 	}
 
+	sc.logger.Infof("shared conn: %p, peerConnection: %p", sc, peerConn)
 	sc.peerConn = peerConn
 	sc.peerConnReady, _, err = rpc.ConfigureForRenegotiation(peerConn, rpc.PeerRoleClient, sc.logger)
 	if err != nil {

robot/web/web.go

@@ -32,6 +32,7 @@ import (
 	"goji.io"
 	"goji.io/pat"
 	googlegrpc "google.golang.org/grpc"
+	"google.golang.org/grpc/metadata"
 
 	"go.viam.com/rdk/config"
 	"go.viam.com/rdk/gostream"
@@ -218,6 +219,18 @@ func (svc *webService) StartModule(ctx context.Context) error {
 		if !strings.HasPrefix(info.FullMethod, "/proto.stream.v1.StreamService") {
 			return handler(ctx, req)
 		}
+		svc.logger.Infof("middleware START")
+		defer svc.logger.Infof("middleware END")
+		crn := metadata.ValueFromIncomingContext(ctx, "calling_resource_name")
+		svc.logger.Infof("NICK!!! crn: %#v", crn)
+		if len(crn) != 1 {
+			return handler(ctx, req)
+		}
+
+		callerResourceName, err := resource.NewFromString(crn[0])
+		if err != nil {
+			return handler(ctx, req)
+		}
 		svc.logger.Infof("ctx before: %#v", ctx)
 		svc.logger.Infof("req: %#v", req)
 		svc.logger.Infof("info: %#v", info)
@@ -245,7 +258,7 @@ func (svc *webService) StartModule(ctx context.Context) error {
 			return handler(ctx, req)
 		}
 		svc.resourceNameToPeerConnectionMu.Lock()
-		pc, ok := svc.resourceNameToPeerConnectionMap[name]
+		pc, ok := svc.resourceNameToPeerConnectionMap[callerResourceName.String()]
 		svc.resourceNameToPeerConnectionMu.Unlock()
 
 		if !ok {
@@ -258,7 +271,7 @@ func (svc *webService) StartModule(ctx context.Context) error {
 			return handler(ctx, req)
 		}
 		ctx = rpc.UnsafeContextWithPeerConnection(ctx, pc)
-		svc.logger.Infof("ctx after: %#v", ctx)
+		svc.logger.Infof("ctx after: %#v, callerResourceName:	%s, pc: %p", ctx, callerResourceName, pc)
 		return handler(ctx, req)
 	}
 	unaryInterceptors = append(unaryInterceptors, f)