server: fix BulkAdd

gripql/schema/schema_test.go

@@ -26,19 +26,16 @@ func TestSchemaScanner(t *testing.T) {
 		t.Fatal(err)
 	}
 	defer func() {
+		kv.Close()
 		os.RemoveAll(conf.KVStorePath)
+		os.RemoveAll(conf.Server.WorkDir)
 	}()
 
 	db := kvgraph.NewKVGraph(kv)
-
 	srv, err := server.NewGripServer(db, conf.Server, nil)
 	if err != nil {
 		t.Fatal(err)
 	}
-	defer func() {
-		kv.Close()
-		os.RemoveAll(conf.Server.WorkDir)
-	}()
 
 	queryClient := gripql.NewQueryDirectClient(srv)
 	editClient := gripql.NewEditDirectClient(srv)

server/api.go

@@ -173,13 +173,13 @@ func (server *GripServer) addEdge(ctx context.Context, elem *gripql.GraphElement
 
 // BulkAdd a stream of inputs and loads them into the graph
 func (server *GripServer) BulkAdd(stream gripql.Edit_BulkAddServer) error {
-
-	var elementStream chan *gripql.GraphElement
 	var graphName string
-
-	wg := &sync.WaitGroup{}
 	var insertCount int32
 	var errorCount int32
+
+	elementStream := make(chan *gripql.GraphElement, 100)
+	wg := &sync.WaitGroup{}
+
 	for {
 		element, err := stream.Recv()
 		if err == io.EOF {
@@ -200,9 +200,8 @@ func (server *GripServer) BulkAdd(stream gripql.Edit_BulkAddServer) error {
 
 		// create a BulkAdd stream per graph
 		// close and switch when a new graph is encountered
-		if element.Graph != graphName && graphName != "" {
+		if element.Graph != graphName {
 			close(elementStream)
-
 			graph, err := server.db.Graph(element.Graph)
 			if err != nil {
 				log.WithFields(log.Fields{"error": err}).Error("BulkAdd: error")