Примерно месяц назад, я рассказал Хабру о проекте Bagri: NoSQL базе данных с открытым кодом, построенной поверх распределенного кэша.

После достаточно неплохого отклика, решил написать статью о том как можно наращивать функционал Bagri путем написания расширений (extensions) используя встроенный API системы.

На данный момент Bagri публикует два API для подключения к внешним системам: DataFormat API и DataStore API.

Первый предназначен для парсинга документов в новом формате и преобразования их во внутренний формат системы, а также для обратного построения документов в новом формате из внутреннего представления системы.

Второй API служит для загрузки/сохранения/удаления документов из внешних систем хранения. Зачастую, для подключения к новому источнику документов необходимо реализовать оба интерфейса.

Я покажу, как реализовать DataStore connector к MongoDB и использовать его в качестве системы хранения документов. В данном случае реализации DataFormat API не требуется, поскольку Mongo предоставляет документы в JSON формате, который изначально поддерживается системой.

Сразу хочу сделать пару замечаний:

Практическая польза от такого коннектора? Очевидно, Mongo можно просто использовать в качестве централизованного хранилища документов. Так же он может быть полезен в сценариях, описанных в данной статье, когда данные уже хранятся в Mongo, но ее возможностей стало не хватать для развития функционала системы;
Я не являюсь знатоком MongoDB, если есть более оптимальные способы работы с ней, я буду рад их услышать;

Итак, начнем.

DataStore API предполагает реализацию интерфейса com.bagri.xdm.cache.api.DocumentStore:

public interface DocumentStore {
	
        /**
         * Lifecycle method. Invoked when the store initialized. 
         * 
         * @param context the environment context
         */
        void init(Map<String, Object> context);
	
        /**
         * Lifecycle method. Invoked when parent schema is closing
         */
        void close();

        /**
         * Load document from persistent store
         * 
         * @param key the document key
         * @return XDM Document instance if corresponding document found, null otherwise
         */
        Document loadDocument(DocumentKey key);

        /**
         * Load bunch of documents from persistent store
         * 
         * @param keys the collection of document keys to load
         * @return the map of loaded documents with their keys
         */
        Map<DocumentKey, Document> loadAllDocuments(Collection<DocumentKey> keys);

        /**
         * Load document keys. Can do it in synch or asynch way.
         * 
         * @return iterator over found document keys
         */
        Iterable<DocumentKey> loadAllDocumentKeys();

        /**
         * Stores document to persistent store.
         * 
         * @param key the document key
         * @param value the XDM document instance
         */
        void storeDocument(DocumentKey key, Document value);

        /**
         * Stores bunch of documents to persistent store
         * 
         * @param entries the map of document keys and corresponding document instances
         */
        void storeAllDocuments(Map<DocumentKey, Document> entries);

        /**
         * Deletes document from persistent store
         * 
         * @param key the document key
         */
        void deleteDocument(DocumentKey key);

        /**
         * Deletes bunch o documents from persistent store 
         * 
         * @param keys the keys identifying documents to be deleted 
         */
        void deleteAllDocuments(Collection<DocumentKey> keys);
		
}

Проще всего будет наследовать наш класс, реализующий интерфейс DocumentStore, от абстрактного класса DocumentStoreBase, предоставляющего набор вспомогательных методов для доступа к контексту системы. Начнем с обработки параметров подключения к внешней системе и его инициализации.

Для подключения к Mongo нам понадобятся адрес сервера mongod, имя базы данных и имена коллекций, из которых мы хотим загружать документы. Определим имена для этих параметров: mongo.db.uri, mongo.db.database, mongo.db.collections. Тогда код инициализации подключения к серверу mongo может выглядеть таким образом:

public class MongoDBStore extends DocumentStoreBase implements DocumentStore {
	
        private MongoClient client;
        private MongoDatabase db;
        private Map<String, MongoCollection<org.bson.Document>> clMap = new HashMap<>();
	
        @Override
        public void init(Map<String, Object> context) {
	    super.setContext(context);
	    String uri = (String) context.get("mongo.db.uri"); 
	    MongoClientURI mcUri = new MongoClientURI(uri);
	    client = new MongoClient(mcUri);

	    String dbName = (String) context.get("mongo.db.database");
	    db = client.getDatabase(dbName);
		
	    String clNames = (String) context.get("mongo.db.collections");
	    boolean all = "*".equals(clNames);
	    List<String> clns = Arrays.asList(clNames.split(","));
	    for (String clName: db.listCollectionNames()) {
		if (all || clns.contains(clName)) {
		    MongoCollection<org.bson.Document> cln = db.getCollection(clName); 
		    clMap.put(clName, cln);
		}
	    }
        }

        @Override
        public void close() {
                client.close();
        }
}

Метод init принимает параметры подключения к серверу mongod из контекста, устанавливает соединение и кэширует объект MongoCollection для каждой коллекции, объявленной для загрузки. Теперь нам нужно реализовать метод для загрузки всех ключей документов.

        private String getMappingKey(String id, String cln) {
	      return id + "::" + cln;
        }
	
        private String[] getMappingParts(String keyMap) {
	      return keyMap.split("::");
        }

        @Override
        public Iterable<DocumentKey> loadAllDocumentKeys() {
	    SchemaRepository repo = getRepository();
            if (repo == null) {
		return null;
	    }

	    String id;
	    DocumentKey key;
	    Map<DocumentKey, String> keyMap = new HashMap<>();
	    for (MongoCollection<org.bson.Document> cln: clMap.values()) {
		String clName = cln.getNamespace().getCollectionName();
		// load _ids only
		MongoCursor<org.bson.Document> cursor =   cln.find().projection(include(“_id”)).iterator();
		
                while (cursor.hasNext()) {
		      org.bson.Document doc = cursor.next();
		      id = doc.get(“_id”).toString();
		      // TODO: handle possible duplicates via revisions
		      key = repo.getFactory().newDocumentKey(id, 0, 1);
		      keyMap.put(key, getMappingKey(id, clName));
		}
	    }
   	    PopulationManagement popManager = repo.getPopulationManagement();
   	    popManager.setKeyMappings(keyMap);
	    return keyMap.keySet();
        }

И методы загрузки документов, соответствующих загруженным ключам:

       @Override
        public Map<DocumentKey, Document> loadAllDocuments(Collection<DocumentKey> keys) {
	    Map<DocumentKey, Document> entries = new HashMap<>(keys.size());
	    for (DocumentKey key: keys) {
		Document doc = loadDocument(key);
		if (doc != null) {
		      entries.put(key, doc);
		}
	    }
	    return entries;
        }

        @Override
        public Document loadDocument(DocumentKey key) {
          SchemaRepository repo = getRepository();
    	  Document doc = null;
	
   	  PopulationManagement popManager = repo.getPopulationManagement(); 
    	  String id = popManager.getKeyMapping(key); 
    	  if (id == null) {
	      return null;
    	  }
    	  String[] mParts = getMappingParts(id);
	  Document newDoc = null;

    	  int[] clns = null;
    	  com.bagri.xdm.system.Collection xcl = repo.getSchema().getCollection(mParts[1]);
    	  if (xcl != null) {
    	        clns = new int[] {xcl.getId()};
    	  }
    	  MongoCollection<org.bson.Document> cln = clMap.get(mParts[1]);

	  Object oid;
    	  Date creDate;
    	  try {
    	        oid = new ObjectId(mParts[0]);
    	        creDate = ((ObjectId) oid).getDate();
    	  } catch (IllegalArgumentException ex) {
	        oid = mParts[0];
    	        creDate = new Date();
    	  }
	  org.bson.Document mongoDoc = cln.find(eq("_id", oid)).first();
	  String content = mongoDoc.toJson(new JsonWriterSettings(true));

   	  try {
       	        DocumentManagementl docMgr = (DocumentManagement) repo.getDocumentManagement(); 
	        newDoc = docMgr.createDocument(key, mParts[0], content, “JSON”, creDate, "owner", 1, clns, true);
	  } catch (XDMException ex) {
	        // TODO: log error, but do not stop the whole loading
	  }
	  return doc;
      }

Для начала этого достаточно, методы storeDocument/storeAllDocuments и deleteDocument/deleteAllDocuments предлагаю реализовать читателю самостоятельно. Также прошу учесть, что приведенный выше код предназначен только в целях демонстрации процесса реализации коннектора и не обрабатывает различные исключительные ситуации и возможные дополнительные параметры конфигурации. Полный код коннектора можно посмотреть и собрать из репозитория bagri-extensions.

Теперь нам нужно зарегистрировать DataStore connector и объявить схему, которая будет его использовать. Для этого нам необходимо добавить конфигурацию коннектора в файл <BAGRI_HOME>/config/config.xml, в секцию dataStores:

<dataStore name="mongo">
    <version>1</version>
    <createdAt>2016-08-01T16:17:20.542+03:00</createdAt>
    <createdBy>admin</createdBy>
    <description>MongoDB data store</description>
    <enabled>true</enabled>
    <storeClass>com.bagri.samples.MongoDBStore</storeClass>
    <properties>
        <entry name="mongo.db.uri">mongodb://localhost:27017</entry>
        <entry name="mongo.db.database">test</entry>
        <entry name="mongo.db.collections">*</entry>
     </properties>
</dataStore>

Мы протестируем работу коннектора на примере коллекции restaurants, рассматриваемой во многих примерах работы с MongoDB. Загрузите коллекцию тестовых документов в Mongo, как показано здесь: docs.mongodb.com/getting-started/shell/import-data. Теперь зарегистрируем схему для работы с MongoDB и настроим ее на загрузку данных из данной коллекции. В том же файле config.xml добавим новую схему в раздел schemas:

Добавляем следующие параметры:

<entry name="xdm.schema.store.enabled">true</entry>
<entry name="xdm.schema.store.type">mongo</entry>
<entry name="mongo.db.collections">restaurants</entry>

Раздел schemas в config.xml с новыми параметрами

<schema name="Mongo" active="true">
            <version>1</version>
            <createdAt>2016-08-01T21:30:58.096+04:00</createdAt>
            <createdBy>admin</createdBy>
            <description>Schema for MongoDB</description>
            <properties>
                <entry name="xdm.schema.store.tx.buffer.size">1024</entry>
                <entry name="xdm.schema.data.backup.read">false</entry>
                <entry name="xdm.schema.trans.backup.async">0</entry>
                <entry name="xdm.schema.store.enabled">true</entry>
                <entry name="xdm.schema.thread.pool">10</entry>
                <entry name="xdm.schema.data.stats.enabled">true</entry>
                <entry name="xdm.schema.query.cache">true</entry>
                <entry name="xdm.schema.store.type">mongo</entry>
                <entry name="mongo.db.collections">restaurants</entry>
                <entry name="xdm.schema.format.default">JSON</entry>
                <entry name="xdm.schema.ports.first">10300</entry>
                <entry name="xdm.schema.ports.last">10400</entry>
                <entry name="xdm.schema.population.size">1</entry>
                <entry name="xdm.schema.population.buffer.size">1000000</entry>
                <entry name="xdm.schema.data.backup.async">1</entry>
                <entry name="xdm.schema.store.data.path">../data/mongo</entry>
                <entry name="xdm.schema.dict.backup.sync">0</entry>
                <entry name="xdm.schema.trans.backup.sync">1</entry>
                <entry name="xdm.schema.query.backup.sync">0</entry>
                <entry name="xdm.schema.buffer.size">16</entry>
                <entry name="xdm.schema.dict.backup.async">1</entry>
                <entry name="xdm.schema.dict.backup.read">true</entry>
                <entry name="xdm.schema.trans.backup.read">false</entry>
                <entry name="xdm.schema.query.backup.async">0</entry>
                <entry name="xdm.schema.members">localhost</entry>
                <entry name="xdm.schema.data.backup.sync">0</entry>
                <entry name="xdm.schema.partition.count">157</entry>
                <entry name="xdm.schema.query.backup.read">true</entry>
	        <entry name="xdm.schema.transaction.timeout">0</entry>
                <entry name="xdm.schema.health.threshold.low">25</entry>
                <entry name="xdm.schema.health.threshold.high">0</entry>
                <entry name="xdm.schema.query.parallel">true</entry>
                <entry name="xdm.schema.partition.pool">32</entry>
                <entry name="xqj.schema.baseUri">file:///../data/mongo/</entry>
                <entry name="xqj.schema.orderingMode">2</entry>
                <entry name="xqj.schema.queryLanguageTypeAndVersion">1</entry>
                <entry name="xqj.schema.bindingMode">0</entry>
                <entry name="xqj.schema.boundarySpacePolicy">1</entry>
                <entry name="xqj.schema.scrollability">1</entry>
                <entry name="xqj.schema.holdability">2</entry>
                <entry name="xqj.schema.copyNamespacesModePreserve">1</entry>
                <entry name="xqj.schema.queryTimeout">0</entry>
                <entry name="xqj.schema.defaultFunctionNamespace"></entry>
                <entry name="xqj.schema.defaultElementTypeNamespace"></entry>
                <entry name="xqj.schema.copyNamespacesModeInherit">1</entry>
                <entry name="xqj.schema.defaultOrderForEmptySequences">2</entry>
                <entry name="xqj.schema.defaultCollationUri"></entry>
                <entry name="xqj.schema.constructionMode">1</entry>
            </properties>
            <collections>
                <collection id="1" name="restaurants">
                    <version>1</version>
                    <createdAt>2016-08-01T01:01:26.965+03:00</createdAt>
                    <createdBy>admin</createdBy>
                    <description>Mongo restaurants collection</description>
                    <enabled>true</enabled>
                </collection>
            </collections>
            <fragments/>
            <indexes/>
            <triggers/>
        </schema>

Кроме того, нужно создать новый файл с профилем сервера. В каталоге <BAGRI_HOME>/config создайте файл mongo.properties и укажите в нем используемую сервером схему:

xdm.cluster.node.schemas=Mongo

Удостоверьтесь, что сервер MongoDB запущен и ждет подключений по адресу, указанному в настройках коннектора. Теперь можно стартовать сервер Bagri. В каталоге <BAGRI_HOME>/bin выполните команду >bgcache.cmd mongo (on Windows) или >./bgcache.sh mongo (on Linux). Данный скрипт стартует одиночный сервер Bagri с настройками из профиля mongo.properties. По окончании загрузки серверный лог должен содержать следующие строчки:

показывающие, что в коннектор инициализировал схему Mongo и загрузил в нее 25359 документов из внешнего сервера MongoDB.

Теперь я покажу, как можно манипулировать документами JSON с помощью запросов XQuery.

Для интерактивного выполнения запросов XQuery нам понадобится клиент, позволяющий это делать. С дистрибутивом Bagri поставляется plugin к VisualVM, предоставляющий данную функциональность. Инструкцию по его установке смотрите здесь.

Запустите административный сервер Bagri <BAGRI_HOME>/bin/bgadmin. Откройте приложение VisualVM, подключитесь к административному серверу Bagri Manager и выберите схему Mongo. Закладка DocumentManagement позволяет работать с документами и коллекциями:

, а закладка QueryManagement с запросами XQuery. Выполните следующий простой запрос для выборки ресторана по его идентификатору:

declare namespace m="http://www.w3.org/2005/xpath-functions/map"; 
let $props := map{'method': 'json', 'indent': fn:true()}

for $uri in fn:uri-collection("restaurants")
let $map := fn:json-doc($uri)
where m:get($map, 'restaurant_id') = '40362098'
return (fn:serialize($map, $props), '\&\#xa;')

* Обратите внимание, что символ перевода строки в самой последней строке экранирован \, так как Хабр превращает его в действительный перевод строки, так что при выполнении запроса символ \ нужно убрать.

Или другой, для выборки ресторанов по типу кухни:

declare namespace m="http://www.w3.org/2005/xpath-functions/map"; 
let $props := map{'method': 'json'}

for $uri in fn:uri-collection("restaurants")
let $map := fn:json-doc($uri)
where m:get($map, 'cuisine') = 'Bakery'
return (fn:serialize($map, $props), '\&\#xa;')

XQuery легко позволяет делать любые выборки, доступные в Mongo (кроме запросов по гео-индексам, прямо из коробки они еще не поддерживаются).

Теперь я покажу запросы, которые не поддерживаются в MongoDB: JOIN. Для этого можно привести данные в коллекции restaurants к более нормализованному виду, например отделить отзывы о ресторанах от данных по самому ресторану и сохранить их в разных коллекциях.

Выполните данный запрос и сохраните результаты в файл, потом сделайте импорт полученных данных в MongoDB, в коллекцию rest-short.

declare namespace m="http://www.w3.org/2005/xpath-functions/map";
let $props := map{'method': 'json'}

for $uri in fn:uri-collection("restaurants")
let $rest := fn:json-doc($uri)
let $rest := m:remove($rest, '_id')
let $rest := m:remove($rest, 'grades')
return (fn:serialize($rest, $props), '\&\#xa;')

Следующий запрос выводит данные по отзывам. Так же сохраните их в отдельный файл и затем импортируйте в MongoDB в коллекцию grades.

declare namespace a="http://www.w3.org/2005/xpath-functions/array";
declare namespace m="http://www.w3.org/2005/xpath-functions/map";
let $props := map{'method': 'json'}

for $uri in fn:uri-collection("restaurants")
let $rest := fn:json-doc($uri)
let $grades := m:get($rest, 'grades')
return 
  for $i in (1 to a:size($grades))
    let $grade := a:get($grades, $i)
    let $date := m:get($grade, 'date')
    return ('{"restaurant_id": "', m:get($rest, 'restaurant_id'), 
	'", "date": ', fn:serialize($date, $props), 
	 ', "grade": "', m:get($grade, 'grade'), 
	'", "score": "', m:get($grade, 'score'), '"}', '\&\#xa;')

Теперь поправьте настройки схемы, чтобы заявить новые коллекции для загрузки:

<schema name="Mongo" active="true">
	     ………...
            <properties>
                <entry name="xdm.schema.store.collections">rest-short, grades</entry>
	         ……...
            </properties>
            <collections>
                <collection id="2" name="rest-short">
                    <version>1</version>
                    <createdAt>2016-08-01T01:01:26.965+03:00</createdAt>
                    <createdBy>admin</createdBy>
                    <description>Restaurant headers collection</description>
                    <enabled>true</enabled>
                </collection>
                <collection id="3" name="grades">
                    <version>1</version>
                    <createdAt>2016-08-01T01:01:26.965+03:00</createdAt>
                    <createdBy>admin</createdBy>
                    <description>Restaurant grades collection</description>
                    <enabled>true</enabled>
                </collection>
            </collections>
            <fragments/>
            <indexes/>
            <triggers/>
</schema>

Рестартуйте сервер Bagri для загрузки новых коллекций с данными. Теперь можно проверить, как работают join’ы. Выполните следующий запрос для формирования полной структуры restaurants из двух коллекций:

declare namespace m="http://www.w3.org/2005/xpath-functions/map";
let $props := map{'method': 'json'}

for $ruri in fn:uri-collection("rest-short")
let $rest := fn:json-doc($ruri)
let $rid := m:get($rest, 'restaurant_id')
let $addr := m:get($rest, 'address')
let $txt := ('{"restaurant_id": "', $rid,
	'", "cuisine": "', m:get($rest, 'cuisine'), 
	'", "name": "', m:get($rest, 'name'), 
	'", "borough": "', m:get($rest, 'borough'), 
	'", "address": ', fn:serialize($addr, $props),
	', "grades": [')
return ($txt, fn:string-join(
  for $guri in fn:uri-collection("grades")
  let $grade := fn:json-doc($guri)
  let $gid := m:get($grade, 'restaurant_id')
  where $gid = $rid
  return fn:serialize(m:remove(m:remove($grade, '_id'), 'restaurant_id'), $props), ', '), ']}\&\#xa;')

Итак, мы с вами рассмотрели как можно реализовать DataStore connector к MongoDB и использовать его в качестве системы хранения документов. Надеюсь эта статья сможет стать для вас отправной точкой для написания других расширений Багри или просто побудит вас более подробно ознакомиться с этим интересным продуктом. Проекту всегда требуются Java разработчки заинтересованные в развитии Bagri, более подробно на код проекта можно посмотреть на Гитхабе.

Как подружить Bagri и MongoDB