return nil, err
	}
	if err = stream.WaitTimeout(createStreamResponseTimeout); err != nil {
		return nil, err
	}

	c.registerStream(stream)
	return stream, nil
}

// registerStream adds the stream s to the connection's list of streams that
// it owns.
func (c *connection) registerStream(s httpstream.Stream) {
	c.streamLock.Lock()
	c.streams[s.Identifier()] = s
	c.streamLock.Unlock()
}

	c := &connection{streams: make(map[uint32]httpstream.Stream)}
	stream0 := &fakeStream{id: 0}
	stream1 := &fakeStream{id: 1}
	stream2 := &fakeStream{id: 2}

	c.registerStream(stream0)
	c.registerStream(stream1)

	if len(c.streams) != 2 {
		t.Fatalf("should have two streams, has %d", len(c.streams))
	}

	// not exists
	c.RemoveStreams(stream2)

	if len(c.streams) != 2 {

		deltaResponsesChan: make(chan *discovery.DeltaDiscoveryResponse, 1),
		stopChan:           make(chan struct{}),
		downstreamDeltas:   downstream,
	}
	p.registerStream(con)
	defer p.unregisterStream(con)

	ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
	defer cancel()

	upstreamConn, err := p.buildUpstreamConn(ctx)
	if err != nil {

	p.connectedMutex.Lock()
	defer p.connectedMutex.Unlock()
	if p.connected != nil && p.connected == c {
		close(p.connected.stopChan)
		p.connected = nil
	}
}

func (p *XdsProxy) registerStream(c *ProxyConnection) {
	p.connectedMutex.Lock()
	defer p.connectedMutex.Unlock()
	if p.connected != nil {
		proxyLog.Warnf("registered overlapping stream; closing previous")
		close(p.connected.stopChan)
	}