implement trickle ice

pkg/impl/signal-server/signaling.go

@@ -96,6 +96,16 @@ func (signalingServer SignalingServer) handleStream(ctx context.Context, name st
 					} else {
 						log.Printf("peers received session answer: %v", received)
 					}
+				case *proto.SignalingMessage_ICECandidate:
+					payload := &bytes.Buffer{}
+					if err := json.NewEncoder(payload).Encode(innerMsg.ICECandidate); err != nil {
+						return err
+					}
+					if received, err := signalingServer.redis.Publish(ctx, signalingServer.redisKeyPrefix+":"+msg.Room+":icecandidate:"+*msg.Receiver, payload.String()).Result(); err != nil {
+						return err
+					} else {
+						log.Printf("peers received candidate: %v: %s", received, innerMsg.ICECandidate.Candidate)
+					}
 				}
 			} else {
 				return err
@@ -111,6 +121,7 @@ func (signalingServer SignalingServer) handleRedisPubSub(ctx context.Context, na
 			signalingServer.redisKeyPrefix+":"+room+":discover:"+name,
 			signalingServer.redisKeyPrefix+":"+room+":offer:"+name,
 			signalingServer.redisKeyPrefix+":"+room+":answer:"+name,
+			signalingServer.redisKeyPrefix+":"+room+":icecandidate:"+name,
 		)
 		defer pubsub.Unsubscribe(ctx)
 		defer pubsub.Close()
@@ -183,6 +194,21 @@ func (signalingServer SignalingServer) handleRedisPubSub(ctx context.Context, na
 					}); err != nil {
 						return err
 					}
+				case signalingServer.redisKeyPrefix + ":" + room + ":icecandidate:" + name:
+					candidate := &proto.ICECandidate{}
+					if err := json.NewDecoder(strings.NewReader(msg.Payload)).Decode(candidate); err != nil {
+						return err
+					}
+					if err := stream.Send(&proto.SignalingMessage{
+						Room:     room,
+						Sender:   candidate.Sender,
+						Receiver: &name,
+						Message: &proto.SignalingMessage_ICECandidate{
+							ICECandidate: candidate,
+						},
+					}); err != nil {
+						return err
+					}
 				}
 			}
 		}