p2p stream improvements

.gitignore

@@ -1,6 +1,9 @@
 **/bin
 **/build
 **/.DS_Store
+**/test.out
+**/profile/cpu
+**/profile/memory
 
 # binaries
 client/build

bringyour/trace.go

@@ -21,7 +21,7 @@ import (
 func IsDoneError(r any) bool {
 	isDoneMessage := func(message string) bool {
 		switch message {
-		case "Done":
+		case "Done", "Done.":
 			return true
 		// pgx
 		case "context canceled", "timeout: context already done: context canceled":

client/client.go

@@ -154,29 +154,35 @@ func encodeUuid(src [16]byte) string {
 	return fmt.Sprintf("%x-%x-%x-%x-%x", src[0:4], src[4:6], src[6:8], src[8:10], src[10:16])
 }
 
-type Path struct {
-	ClientId *Id
-	StreamId *Id
+type TransferPath struct {
+	SourceId      *Id
+	DestinationId *Id
+	StreamId      *Id
 }
 
-func NewPath(clientId *Id, streamId *Id) *Path {
-	return &Path{
-		ClientId: clientId,
-		StreamId: streamId,
+func NewTransferPath(sourceId *Id, destinationId *Id, streamId *Id) *TransferPath {
+	return &TransferPath{
+		SourceId:      sourceId,
+		DestinationId: destinationId,
+		StreamId:      streamId,
 	}
 }
 
-func fromConnectPath(path connect.Path) *Path {
-	return &Path{
-		ClientId: newId(path.ClientId),
-		StreamId: newId(path.StreamId),
+func fromConnect(path connect.TransferPath) *TransferPath {
+	return &TransferPath{
+		SourceId:      newId(path.SourceId),
+		DestinationId: newId(path.DestinationId),
+		StreamId:      newId(path.StreamId),
 	}
 }
 
-func (self *Path) toConnectPath() connect.Path {
-	path := connect.Path{}
-	if self.ClientId != nil {
-		path.ClientId = connect.Id(self.ClientId.id)
+func (self *TransferPath) toConnect() connect.TransferPath {
+	path := connect.TransferPath{}
+	if self.SourceId != nil {
+		path.SourceId = connect.Id(self.SourceId.id)
+	}
+	if self.DestinationId != nil {
+		path.DestinationId = connect.Id(self.DestinationId.id)
 	}
 	if self.StreamId != nil {
 		path.StreamId = connect.Id(self.StreamId.id)

client/device.go

@@ -251,7 +251,7 @@ func (self *BringYourDevice) connectChanged(connectEnabled bool) {
 }
 
 // `ReceivePacketFunction`
-func (self *BringYourDevice) receive(source connect.Path, ipProtocol connect.IpProtocol, packet []byte) {
+func (self *BringYourDevice) receive(source connect.TransferPath, ipProtocol connect.IpProtocol, packet []byte) {
 	// deviceLog("GOT A PACKET %d", len(packet))
 	for _, receiveCallback := range self.receiveCallbacks.Get() {
 		receiveCallback(source, ipProtocol, packet)
@@ -357,20 +357,20 @@ func (self *BringYourDevice) SetDestination(specs *ProviderSpecList, provideMode
 				connectSpecs = append(connectSpecs, specs.Get(i).toConnectProviderSpec())
 			}
 
-			paths := []connect.Path{}
+			destinations := []connect.MultiHopId{}
 			for _, connectSpec := range connectSpecs {
 				if connectSpec.ClientId != nil {
-					paths = append(paths, connect.Path{ClientId: *connectSpec.ClientId})
+					destinations = append(destinations, connect.RequireMultiHopId(*connectSpec.ClientId))
 				}
 			}
 
 			// connect to a single client
 			// no need to optimize this case, use the simplest user nat client
-			if DebugUseSingleClientConnect && len(connectSpecs) == len(paths) && len(paths) == 1 {
+			if DebugUseSingleClientConnect && len(connectSpecs) == len(destinations) && len(destinations) == 1 {
 				self.remoteUserNatClient, returnErr = connect.NewRemoteUserNatClient(
 					self.client,
 					self.receive,
-					paths,
+					destinations,
 					protocol.ProvideMode_Network,
 				)
 				if returnErr != nil {
@@ -419,7 +419,7 @@ func (self *BringYourDevice) Shuffle() {
 func (self *BringYourDevice) SendPacket(packet []byte, n int32) bool {
 	packetCopy := make([]byte, n)
 	copy(packetCopy, packet[0:n])
-	source := connect.Path{ClientId: self.clientId}
+	source := connect.SourceId(self.clientId)
 
 	self.stateLock.Lock()
 	remoteUserNatClient := self.remoteUserNatClient
@@ -445,7 +445,7 @@ func (self *BringYourDevice) SendPacket(packet []byte, n int32) bool {
 }
 
 func (self *BringYourDevice) AddReceivePacket(receivePacket ReceivePacket) Sub {
-	receive := func(destination connect.Path, ipProtocol connect.IpProtocol, packet []byte) {
+	receive := func(destination connect.TransferPath, ipProtocol connect.IpProtocol, packet []byte) {
 		receivePacket.ReceivePacket(packet)
 	}
 	callbackId := self.receiveCallbacks.Add(receive)

client/gomobile.go

@@ -83,13 +83,13 @@ func NewIntList() *IntList {
 	}
 }
 
-type PathList struct {
-	exportedList[*Path]
+type TransferPathList struct {
+	exportedList[*TransferPath]
 }
 
-func NewPathList() *PathList {
-	return &PathList{
-		exportedList: *newExportedList[*Path](),
+func NewTransferPathList() *TransferPathList {
+	return &TransferPathList{
+		exportedList: *newExportedList[*TransferPath](),
 	}
 }