//     Data Access is a Java library to store data
//     Copyright (C) 2018 Adrián Romero Corchado.
//
//     This file is part of Data Access
//
//     Licensed under the Apache License, Version 2.0 (the "License");
//     you may not use this file except in compliance with the License.
//     You may obtain a copy of the License at
//     
//         http://www.apache.org/licenses/LICENSE-2.0
//     
//     Unless required by applicable law or agreed to in writing, software
//     distributed under the License is distributed on an "AS IS" BASIS,
//     WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//     See the License for the specific language governing permissions and
//     limitations under the License.
//
package com.adr.data.mongo;

import com.adr.data.DataException;
import com.adr.data.DataLink;
import com.adr.data.record.Header;
import com.adr.data.record.Record;
import com.adr.data.record.Records;
import com.adr.data.var.Parameters;
import com.adr.data.var.Variant;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoDatabase;
import com.mongodb.client.model.Filters;
import com.mongodb.client.model.ReplaceOptions;
import java.util.ArrayList;

import java.util.List;
import org.bson.Document;
import org.bson.conversions.Bson;

/**
 *
 * @author adrian
 */
public class MongoDataLink implements DataLink {

    private final static ReplaceOptions UPSERT = new ReplaceOptions().upsert(true);

    private final MongoDatabase database;

    public MongoDataLink(MongoDatabase database) {
        this.database = database;
    }

    @Override
    public void execute(Header headers, List<Record> records) throws DataException {

        for (Record r : records) {
            String entity = Records.getCollection(r);
            MongoCollection<Document> collection = database.getCollection(entity);

            Document doc = new Document();
            List<Bson> filters = new ArrayList<>();
            for (String f : r.getNames()) {
                String realname;
                if (!f.equals("COLLECTION.KEY") && !f.contains("..")) {
                    Variant v = r.get(f);
                    if (f.endsWith(".KEY")) {
                        realname = f.substring(0, f.length() - 4);
                        filters.add(Filters.eq(realname, v.asObject()));
                    } else {
                        realname = f;
                    }
                    Parameters p = new DocumentParameters(doc, realname);
                    v.getKind().write(p, v);
                }
            }
            collection.replaceOne(Filters.and(filters), doc, UPSERT);
        }
    }
}

@@ -1870,17 +1870,16 @@
 ntity);%0A
-%0A
         
@@ -1886,38 +1886,8 @@
     
-Document doc = new Document();
 %0A   
@@ -1935,16 +1935,67 @@
 st%3C%3E();%0A
+            if (Records.isDeleteSentence(r)) %7B%0A    
         
@@ -2042,24 +2042,28 @@
             
+    
 String realn
@@ -2067,16 +2067,20 @@
 alname;%0A
+    
         
@@ -2163,16 +2163,20 @@
         
+    
 Variant 
@@ -2189,16 +2189,20 @@
 get(f);%0A
+    
         
@@ -2235,24 +2235,28 @@
 (%22.KEY%22)) %7B%0A
+    
             
@@ -2334,16 +2334,20 @@
         
+    
 filters.
@@ -2411,77 +2411,720 @@
     
-%7D else %7B%0A                        realname = f;%0A                    %7D%0A
+    %7D%0A                    %7D%0A                %7D                %0A                collection.deleteOne(Filters.and(filters));%0A            %7D else %7B%0A                Document doc = new Document();            %0A                for (String f : r.getNames()) %7B%0A                    String realname;%0A                    if (!f.equals(%22COLLECTION.KEY%22) && !f.contains(%22..%22)) %7B%0A                        Variant v = r.get(f);%0A                        if (f.endsWith(%22.KEY%22)) %7B%0A                            realname = f.substring(0, f.length() - 4);%0A                            filters.add(Filters.eq(realname, v.asObject()));%0A                        %7D else %7B%0A                            realname = f;%0A                        %7D%0A    
     
@@ -3213,16 +3213,20 @@
         
+    
 v.getKin
@@ -3242,16 +3242,20 @@
 (p, v);%0A
+    
         
@@ -3268,34 +3268,42 @@
   %7D%0A            
-%7D%0A
+    %7D%0A    
             coll
@@ -3356,16 +3356,31 @@
 PSERT);%0A
+            %7D %0A
         

Implement deleting documents