Revert "Revert "add bidirectional streaming support for external snapshot (#276)

Author: mangalaman93

client_test.go

@@ -35,7 +35,7 @@ func TestOpen(t *testing.T) {
 	require.ErrorContains(t, err, "invalid connection string: host url must have both host and port")
 
 	_, err = dgo.Open("dgraph://localhost:")
-	require.ErrorContains(t, err, "missing port after port-separator colon")
+	require.ErrorContains(t, err, "invalid connection string: missing port after port-separator colon")
 
 	_, err = dgo.Open("dgraph://localhost:9180?sslmode=verify-ca")
 	require.ErrorContains(t, err, "first record does not look like a TLS handshake")
@@ -77,7 +77,7 @@ func TestOpen(t *testing.T) {
 	require.ErrorContains(t, err, "invalid namespace ID: strconv.ParseUint: parsing \"root\": invalid syntax")
 
 	_, err = dgo.Open("dgraph://user:pass@localhost:9180?namespace=1")
-	require.NoError(t, err)
+	require.ErrorContains(t, err, "invalid username or password")
 
 	_, err = dgo.Open("dgraph://groot:password@localhost:9180")
 	require.NoError(t, err)

open.go

@@ -169,9 +169,12 @@ func Open(connStr string) (*Dgraph, error) {
 	if apiKey != "" && bearerToken != "" {
 		return nil, errors.New("invalid connection string: both apikey and bearertoken cannot be provided")
 	}
-	if !strings.Contains(u.Host, ":") {
+	if len(strings.Split(u.Host, ":")) != 2 {
 		return nil, errors.New("invalid connection string: host url must have both host and port")
 	}
+	if strings.Split(u.Host, ":")[1] == "" {
+		return nil, errors.New("invalid connection string: missing port after port-separator colon")
+	}
 
 	opts := []ClientOption{}
 	if apiKey != "" {
@@ -257,6 +260,11 @@ func NewRoundRobinClient(endpoints []string, opts ...ClientOption) (*Dgraph, err
 		}
 	}
 
+	if _, err := dc[0].CheckVersion(context.Background(), &api.Check{}); err != nil {
+		d.Close()
+		return nil, fmt.Errorf("failed to ping: %w", err)
+	}
+
 	return d, nil
 }
 

protos/api.proto

@@ -29,7 +29,7 @@ service Dgraph {
 	rpc AllocateIDs(AllocateIDsRequest) returns (AllocateIDsResponse) {}
 
 	rpc UpdateExtSnapshotStreamingState(UpdateExtSnapshotStreamingStateRequest) returns (UpdateExtSnapshotStreamingStateResponse) {}
-	rpc StreamExtSnapshot(stream StreamExtSnapshotRequest) returns (StreamExtSnapshotResponse) {}
+	rpc StreamExtSnapshot(stream StreamExtSnapshotRequest) returns (stream StreamExtSnapshotResponse) {}
 
 	rpc CreateNamespace(CreateNamespaceRequest) returns (CreateNamespaceResponse) {}
 	rpc DropNamespace(DropNamespaceRequest) returns (DropNamespaceResponse) {}
@@ -264,7 +264,9 @@ message StreamExtSnapshotRequest {
   StreamPacket pkt = 3;
 }
 
-message StreamExtSnapshotResponse {}
+message StreamExtSnapshotResponse {
+  bool finish = 1;
+}
 
 message StreamPacket {
   bytes data = 1;