return &Connection{
		Connection: xds.NewConnection(peerAddr, stream),
	}
}

func (conn *Connection) Initialize(node *core.Node) error {
	return conn.s.initConnection(node, conn, conn.ids)
}

func (conn *Connection) Close() {
	conn.s.closeConnection(conn)
}

func (conn *Connection) Watcher() xds.Watcher {
	return conn.proxy
}

				continue
			}
			firstRequest = false
			if req.Node == nil || req.Node.Id == "" {
				con.ErrorCh() <- status.New(codes.InvalidArgument, "missing node information").Err()
				return
			}
			if err := s.initConnection(req.Node, con, identities); err != nil {
				con.ErrorCh() <- err
				return
			}
			defer s.closeConnection(con)
			deltaLog.Infof("ADS: new delta connection for node:%s", con.ID())
		}

		select {