bug msg"="could not negotiate" "error"="ICEAgent does not exist: unable to restart ICETransport

[brucel11qwe]

2024/05/15 19:34:47 "level"=0 "msg"="resuming connection..." "reconnectCount"=0
2024/05/15 19:34:50 "msg"="could not negotiate" "error"="ICEAgent does not exist: unable to restart ICETransport"
2024/05/15 19:34:50 "msg"="resume connection failed" "error"="ICEAgent does not exist: unable to restart ICETransport"
2024/05/15 19:34:50 "level"=0 "msg"="resuming connection..." "reconnectCount"=1
2024/05/15 19:34:58 "msg"="could not negotiate" "error"="ICEAgent does not exist: unable to restart ICETransport"
2024/05/15 19:34:58 "msg"="resume connection failed" "error"="ICEAgent does not exist: unable to restart ICETransport"
2024/05/15 19:34:58 "level"=0 "msg"="resuming connection..." "reconnectCount"=2
2024/05/15 19:35:06 "msg"="could not negotiate" "error"="ICEAgent does not exist: unable to restart ICETransport"
2024/05/15 19:35:06 "msg"="resume connection failed" "error"="ICEAgent does not exist: unable to restart ICETransport"
2024/05/15 19:35:07 "level"=0 "msg"="resuming connection..." "reconnectCount"=3
2024/05/15 19:35:14 "msg"="could not negotiate" "error"="ICEAgent does not exist: unable to restart ICETransport"
2024/05/15 19:35:14 "msg"="resume connection failed" "error"="ICEAgent does not exist: unable to restart ICETransport"
2024/05/15 19:35:17 "level"=0 "msg"="resuming connection..." "reconnectCount"=4
2024/05/15 19:35:22 "msg"="could not negotiate" "error"="ICEAgent does not exist: unable to restart ICETransport"
2024/05/15 19:35:22 "msg"="resume connection failed" "error"="ICEAgent does not exist: unable to restart ICETransport"
2024/05/15 19:35:26 "level"=0 "msg"="resuming connection..." "reconnectCount"=5
2024/05/15 19:35:29 "msg"="could not negotiate" "error"="ICEAgent does not exist: unable to restart ICETransport"
2024/05/15 19:35:29 "msg"="resume connection failed" "error"="ICEAgent does not exist: unable to restart ICETransport"
2024/05/15 19:35:37 "level"=0 "msg"="resuming connection..." "reconnectCount"=6
2024/05/15 19:35:37 "msg"="could not negotiate" "error"="ICEAgent does not exist: unable to restart ICETransport"
2024/05/15 19:35:37 "msg"="resume connection failed" "error"="ICEAgent does not exist: unable to restart ICETransport"
2024/05/15 19:35:48 "level"=0 "msg"="resuming connection..." "reconnectCount"=7
2024/05/15 19:35:48 "msg"="could not negotiate" "error"="ICEAgent does not exist: unable to restart ICETransport"
2024/05/15 19:35:48 "msg"="resume connection failed" "error"="ICEAgent does not exist: unable to restart ICETransport"
2024/05/15 19:36:03 "level"=0 "msg"="resuming connection..." "reconnectCount"=8
2024/05/15 19:36:03 "msg"="could not negotiate" "error"="ICEAgent does not exist: unable to restart ICETransport"
2024/05/15 19:36:03 "msg"="resume connection failed" "error"="ICEAgent does not exist: unable to restart ICETransport"
2024/05/15 19:36:22 "level"=0 "msg"="resuming connection..." "reconnectCount"=9
2024/05/15 19:36:22 "msg"="could not negotiate" "error"="ICEAgent does not exist: unable to restart ICETransport"
2024/05/15 19:36:22 "msg"="resume connection failed" "error"="ICEAgent does not exist: unable to restart ICETransport"

`package main

import (
"fmt"
"github.com/google/gopacket"
"github.com/google/gopacket/layers"
lksdk "github.com/livekit/server-sdk-go/v2"
"github.com/livekit/server-sdk-go/v2/common"
"github.com/pkg/errors"
"github.com/vishvananda/netlink"
"gopkg.in/yaml.v2"
"io"
"os"
"os/signal"
"strings"
"syscall"
)

var (
device io.ReadWriteCloser
topic string
cidr string
mtu int
dst string
identity string
tunName string
prefix string
)

type Config struct {
LiveKit struct {
Host string yaml:"host"
APIKey string yaml:"apiKey"
APISecret string yaml:"apiSecret"
RoomName string yaml:"roomName"
} yaml:"livekit"
Participant struct {
Identity string yaml:"identity"
TunName string yaml:"tunName"
MTU int yaml:"mtu"
CIDR string yaml:"cidr"
Topic string yaml:"topic"
Dst string yaml:"dst"
} yaml:"participant"
}

func main() {

dir, _ := os.Getwd()
path := fmt.Sprintf("%s/%s", dir, "config.yaml")

file, err := os.ReadFile(path)
if err != nil {
	panic(err)

}
t := Config{}
err = yaml.Unmarshal(file, &t)
if err != nil {
	panic(err)

}
mtu = t.Participant.MTU
tunName = t.Participant.TunName
cidr = t.Participant.CIDR
topic = t.Participant.Topic
dst = t.Participant.Dst
identity = t.Participant.Identity
prefix = getPrefix(cidr)

//创建tun设备
device, _, err = common.CreateTunDevice(tunName, mtu)
if err != nil {
	panic(err)

}
err = prepareInterface(&tunConfig{
	InterfaceName:    tunName,
	InterfaceAddress: cidr,
	InterfaceMTU:     mtu,
})
if err != nil {
	panic(err)

}

room, err := lksdk.ConnectToRoom(t.LiveKit.Host, lksdk.ConnectInfo{
	APIKey:              t.LiveKit.APIKey,
	APISecret:           t.LiveKit.APISecret,
	RoomName:            t.LiveKit.RoomName,
	ParticipantIdentity: identity,
}, &lksdk.RoomCallback{
	ParticipantCallback: lksdk.ParticipantCallback{
		OnDataPacket: onDataPacket,
	},
	OnDisconnected: func() {
		fmt.Println("disconnected")

	},
})
if err != nil {
	panic(err)
}

//传package
packets := make(chan []byte, mtu)
go func() {
	for v := range packets {
		data, err1 := filterPacket(v, prefix)
		if err1 != nil {
			fmt.Println("Error handling packet:", err1)
			continue
		}
		err1 = room.LocalParticipant.PublishDataPacket(&lksdk.UserDataPacket{
			Payload: data,
			Topic:   topic,
		}, lksdk.WithDataPublishReliable(true),
			lksdk.WithDataPublishDestination([]string{dst}))

		if err1 != nil {
			fmt.Println("error publishing data", err1)
			continue
		}
	}
}()

//tun流量读取
go func() {
	for {
		frame := make([]byte, mtu)
		_, err1 := device.Read(frame)
		if err1 != nil {
			fmt.Println("Error reading from TUN device:", err)
			continue
		}

		packets <- frame
	}
}()

sigs := make(chan os.Signal, 1)
signal.Notify(sigs, syscall.SIGHUP, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT)
<-sigs
room.Disconnect()

}

func filterPacket(packet []byte, prefix string) ([]byte, error) {
var ip layers.IPv4

if err := ip.DecodeFromBytes(packet, gopacket.NilDecodeFeedback); err != nil {

	return nil, errors.Wrap(err, "could not parse IP packet")
}
dst1 := ip.DstIP.String()

if !strings.Contains(dst1, prefix) {
	return nil, errors.New("invalid destination")
}
//改成取cidr前2位

fmt.Println("dst1", dst1)

return packet, nil

}

// 收到的包都写入网卡里面
func onDataPacket(data lksdk.DataPacket, params lksdk.DataReceiveParams) {
if data.ToProto().GetUser().GetTopic() == topic {
a := data.ToProto().GetUser().GetPayload()
if device == nil {
fmt.Println("TUN device not initialized")
return
}

	_, err := device.Write(a)
	if err != nil {
		fmt.Println("Error writing to TUN device:", err)
	}
}

}

type tunConfig struct {
InterfaceName string
InterfaceAddress string
InterfaceMTU int
}

func prepareInterface(c *tunConfig) error {
link, err := netlink.LinkByName(c.InterfaceName)
if err != nil {
return err
}

addr, err := netlink.ParseAddr(c.InterfaceAddress)
if err != nil {
	return err
}

err = netlink.LinkSetMTU(link, c.InterfaceMTU)
if err != nil {
	return err
}

err = netlink.AddrAdd(link, addr)
if err != nil {
	return err
}

err = netlink.LinkSetUp(link)
if err != nil {
	return err
}
return nil

}

func getPrefix(cidr string) string {

ipParts := strings.Split(cidr, ".")

return strings.Join(ipParts[:2], ".")

}
`
livekit-server v1.6.1
docker run --rm
-p 7880:7880
-p 7881:7881
-p 7882:7882/udp
-v $PWD/livekit.yaml:/livekit.yaml
livekit/livekit-server
--config /livekit.yaml
--node-ip=xxxx