allow queries to wait for peers to connect within its timeout

request.go

@@ -192,10 +192,6 @@ func (m *RequestManager) Query(
 	timeout time.Duration,
 	opts ...RequestOption,
 ) ([]ReqEvent, *ReqClosed, error) {
-	if !m.envoy.IsConnected() {
-		return nil, nil, nil
-	}
-
 	id, eventsCh, closedCh, err := m.newStream(filters, true, opts...)
 	if err != nil {
 		return nil, nil, err
@@ -295,10 +291,8 @@ func (m *RequestManager) handleEvents() {
 			switch ev.Kind {
 			case EventConnected:
 				for _, req := range m.reqs {
-					if !req.isQuery {
 					m.activate(req)
 				}
-				}
 			case EventDisconnected:
 				for _, req := range m.reqs {
 					req.active = false

request_test.go

@@ -485,17 +485,40 @@ func TestRequestManager_Query(t *testing.T) {
 		assert.Nil(t, closed)
 	})
 
-	t.Run("returns nil nil when disconnected", func(t *testing.T) {
+	t.Run("connects within timeout and returns events", func(t *testing.T) {
-		_, envoy := newMockEnvoy(t)
+		p, envoy := newMockEnvoy(t)
-		// do not connect
+		// do NOT connect
 
 		m := NewRequestManager(envoy)
 		t.Cleanup(func() { m.Close() })
 
-		events, closed, err := m.Query([][]byte{[]byte(`{}`)}, TestTimeout)
+		filters := [][]byte{[]byte(`{}`)}
-		assert.NoError(t, err)
+		eventData := []byte(`{"id":"abc"}`)
 
-		assert.Nil(t, events)
+		go func() {
+			// wait to connect
+			time.Sleep(50 * time.Millisecond)
+			p.connect()
+
+			// listen for REQ on pool side to extract subscription id
+			reqBytes := <-p.sent
+			subID, _, err := envelope.FindReq(reqBytes)
+			if err != nil {
+				t.Errorf("FindReq: %v", err)
+				return
+			}
+
+			// send event and eose
+			p.receive(envelope.EncloseSubscriptionEvent(subID, eventData))
+			p.receive(envelope.EncloseEOSE(subID))
+		}()
+
+		// start the query while the peer is disconnected
+		// because it connects within the timeout, the query should still
+		// return events
+		events, closed, err := m.Query(filters, TestTimeout)
+		assert.NoError(t, err)
+		assert.Len(t, events, 1)
 		assert.Nil(t, closed)
 	})
 }