PublicRepository.java

  1. /*
  2.  * Copyright (c) 2007-2017 MetaSolutions AB
  3.  *
  4.  * Licensed under the Apache License, Version 2.0 (the "License");
  5.  * you may not use this file except in compliance with the License.
  6.  * You may obtain a copy of the License at
  7.  *
  8.  *     http://www.apache.org/licenses/LICENSE-2.0
  9.  *
  10.  * Unless required by applicable law or agreed to in writing, software
  11.  * distributed under the License is distributed on an "AS IS" BASIS,
  12.  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13.  * See the License for the specific language governing permissions and
  14.  * limitations under the License.
  15.  */

  17. package org.entrystore.impl;

  19. import com.github.benmanes.caffeine.cache.Cache;
  20. import com.github.benmanes.caffeine.cache.Caffeine;
  21. import com.google.common.collect.Queues;
  22. import org.eclipse.rdf4j.model.IRI;
  23. import org.eclipse.rdf4j.model.Model;
  24. import org.eclipse.rdf4j.model.Resource;
  25. import org.eclipse.rdf4j.model.Value;
  26. import org.eclipse.rdf4j.model.ValueFactory;
  27. import org.eclipse.rdf4j.repository.Repository;
  28. import org.eclipse.rdf4j.repository.RepositoryConnection;
  29. import org.eclipse.rdf4j.repository.RepositoryException;
  30. import org.eclipse.rdf4j.repository.sail.SailRepository;
  31. import org.eclipse.rdf4j.sail.memory.MemoryStore;
  32. import org.eclipse.rdf4j.sail.nativerdf.NativeStore;
  33. import org.entrystore.AuthorizationException;
  34. import org.entrystore.Context;
  35. import org.entrystore.ContextManager;
  36. import org.entrystore.Entry;
  37. import org.entrystore.EntryType;
  38. import org.entrystore.GraphType;
  39. import org.entrystore.PrincipalManager;
  40. import org.entrystore.config.Config;
  41. import org.entrystore.repository.RepositoryManager;
  42. import org.entrystore.repository.config.Settings;
  43. import org.slf4j.Logger;
  44. import org.slf4j.LoggerFactory;

  46. import java.io.File;
  47. import java.net.URI;
  48. import java.util.Date;
  49. import java.util.HashSet;
  50. import java.util.Iterator;
  51. import java.util.Queue;
  52. import java.util.Set;
  53. import java.util.concurrent.ConcurrentMap;

  55. /**
  56.  * @author Hannes Ebner
  57.  */
  58. public class PublicRepository {
  59.     
  60.     Logger log = LoggerFactory.getLogger(PublicRepository.class);
  61.     
  62.     private boolean rebuilding = false;
  63.     
  64.     private Repository repository;
  65.     
  66.     private RepositoryManager rm;
  67.     
  68.     private PrincipalManager pm;

  70.     private Thread entrySubmitter;

  72.     private final Cache<URI, Entry> postQueue = Caffeine.newBuilder().build();

  74.     private final Queue<Entry> deleteQueue = Queues.newConcurrentLinkedQueue();

  76.     private static final int BATCH_SIZE = 1000;

  78.     public class EntrySubmitter extends Thread {

  80.         @Override
  81.         public void run() {
  82.             while (!interrupted()) {
  83.                 postQueue.cleanUp();
  84.                 int batchCount = 0;

  86.                 if (postQueue.estimatedSize() > 0 || !deleteQueue.isEmpty()) {
  87.                     if (!deleteQueue.isEmpty()) {
  88.                         Set<Entry> entriesToRemove = new HashSet<>();
  89.                         synchronized (deleteQueue) {
  90.                             while (batchCount < BATCH_SIZE) {
  91.                                 Entry e = deleteQueue.poll();
  92.                                 if (e == null) {
  93.                                     break;
  94.                                 }
  95.                                 entriesToRemove.add(e);
  96.                                 batchCount++;
  97.                             }
  98.                         }
  99.                         if (batchCount > 0) {
  100.                             log.info("Removing " + batchCount + " entries from Public Repository, " + deleteQueue.size() + " entries remaining in removal queue");
  101.                             removeEntries(entriesToRemove);
  102.                         }
  103.                     }
  104.                     if (postQueue.estimatedSize() > 0) {
  105.                         Set<Entry> entriesToUpdate = new HashSet<>();
  106.                         synchronized (postQueue) {
  107.                             ConcurrentMap<URI, Entry> postQueueMap = postQueue.asMap();
  108.                             Iterator<URI> it = postQueueMap.keySet().iterator();
  109.                             while (batchCount < BATCH_SIZE && it.hasNext()) {
  110.                                 URI key = it.next();
  111.                                 Entry entry = postQueueMap.get(key);
  112.                                 postQueueMap.remove(key, entry);
  113.                                 if (entry == null) {
  114.                                     log.warn("Value for key " + key + " is null in Public Repository submit queue");
  115.                                 }
  116.                                 entriesToUpdate.add(entry);
  117.                                 batchCount++;
  118.                             }
  119.                         }
  120.                         postQueue.cleanUp();
  121.                         log.info("Sending " + entriesToUpdate.size() + " entries for update in Public Repository, " + postQueue.estimatedSize() + " entries remaining in post queue");
  122.                         updateEntries(entriesToUpdate);
  123.                     }
  124.                 } else {
  125.                     try {
  126.                         Thread.sleep(10000);
  127.                     } catch (InterruptedException ie) {
  128.                         log.info("Public Repository submitter got interrupted, shutting down submitter thread");
  129.                         return;
  130.                     }
  131.                 }
  132.             }
  133.         }

  135.     }

  137.     public PublicRepository(RepositoryManager rm) {
  138.         this.rm = rm;
  139.         this.pm = rm.getPrincipalManager();
  140.         Config config = rm.getConfiguration();

  142.         String storeType = config.getString(Settings.REPOSITORY_PUBLIC_TYPE, "memory").trim();
  143.         log.info("Public repository type: " + storeType);
  144.         
  145.         if (storeType.equalsIgnoreCase("memory")) {
  146.             this.repository = new SailRepository(new MemoryStore());
  147.         } else if (storeType.equalsIgnoreCase("native")) {
  148.             if (!config.containsKey(Settings.REPOSITORY_PUBLIC_PATH)) {
  149.                 log.error("Incomplete configuration of public repository");
  150.             } else {
  151.                 File path = new File(config.getURI(Settings.REPOSITORY_PUBLIC_PATH));
  152.                 String indexes = config.getString(Settings.REPOSITORY_PUBLIC_INDEXES);
  153.                 ((RepositoryManagerImpl) rm).checkAndUpgradeNativeStore(path, indexes);
  154.                 log.info("Public repository: using Native Store at {} with indexes {}", path, indexes);
  155.                 
  156.                 NativeStore store = null;
  157.                 if (indexes != null) {
  158.                     store = new NativeStore(path, indexes);
  159.                 } else {
  160.                     store = new NativeStore(path);
  161.                 }
  162.                 if (store != null) {
  163.                     this.repository = new SailRepository(store);
  164.                 }
  165.             }
  166.         }
  167.         
  168.         if (this.repository == null) {
  169.             log.error("Failed to create public repository");
  170.             return;
  171.         }

  173.         try {
  174.             repository.init();
  175.         } catch (RepositoryException e) {
  176.             log.error(e.getMessage());
  177.         }

  179.         if (getTripleCount() == 0 ||
  180.                 "on".equalsIgnoreCase(config.getString(Settings.REPOSITORY_PUBLIC_REBUILD_ON_STARTUP, "off"))) {
  181.             rebuildRepository();
  182.         }

  184.         entrySubmitter = new PublicRepository.EntrySubmitter();
  185.         entrySubmitter.start();
  186.     }
  187.     
  188.     public RepositoryConnection getConnection() {
  189.         try {
  190.             return repository.getConnection();
  191.         } catch (RepositoryException e) {
  192.             log.error(e.getMessage());
  193.         }
  194.         return null;
  195.     }

  197.     public void enqueue(Entry entry) {
  198.         URI entryURI = entry.getEntryURI();
  199.         synchronized (postQueue) {
  200.             log.info("Adding document to update queue: " + entryURI);
  201.             postQueue.put(entryURI, entry);
  202.         }
  203.     }

  205.     public void remove(Entry entry) {
  206.         URI entryURI = entry.getEntryURI();
  207.         synchronized (deleteQueue) {
  208.             log.info("Adding entry to delete queue: " + entryURI);
  209.             deleteQueue.add(entry);
  210.         }
  211.     }

  213.     private void addEntry(Entry e, RepositoryConnection rc) throws RepositoryException {
  214.         if (isAdministrative(e)) {
  215.             return;
  216.         }
  217.         URI currentUser = pm.getAuthenticatedUserURI();
  218.         try {
  219.             pm.setAuthenticatedUserURI(pm.getGuestUser().getURI());
  220.             try {
  221.                 ValueFactory vf = repository.getValueFactory();
  222.                 IRI contextURI = vf.createIRI(e.getContext().getURI().toString());

  224.                 // entry
  225.                 /* DEACTIVATED
  226.                 Graph entryGraph = e.getGraph();
  227.                 URI entryNG = vf.createIRI(e.getEntryURI().toString());
  228.                 if (entryGraph != null) {
  229.                     rc.add(entryGraph, entryNG, contextURI);
  230.                 }
  231.                 */

  233.                 // metadata
  234.                 Model mdGraph = null;
  235.                 IRI mdNG = null;
  236.                 if (e.getLocalMetadata() != null) {
  237.                     mdGraph = e.getLocalMetadata().getGraph();
  238.                     mdNG = vf.createIRI(e.getLocalMetadataURI().toString());
  239.                 }

  241.                 // ext metadata
  242.                 Model extMdGraph = null;
  243.                 IRI extMdNG = null;
  244.                 if (e.getCachedExternalMetadata() != null) {
  245.                     extMdGraph = e.getCachedExternalMetadata().getGraph();
  246.                     extMdNG = vf.createIRI(e.getCachedExternalMetadataURI().toString());
  247.                 }

  249.                 // resource
  250.                 Model resGraph = null;
  251.                 IRI resNG = null;
  252.                 if (GraphType.Graph.equals(e.getGraphType()) && EntryType.Local.equals(e.getEntryType())) {
  253.                     resGraph = (Model) e.getResource();
  254.                     resNG = vf.createIRI(e.getResourceURI().toString());
  255.                 }

  257.                 if (mdGraph != null) {
  258.                     rc.add(mdGraph, mdNG, contextURI);
  259.                 }
  260.                 if (extMdGraph != null) {
  261.                     rc.add(extMdGraph, extMdNG, contextURI);
  262.                 }
  263.                 if (resGraph != null) {
  264.                     rc.add(resGraph, resNG, contextURI);
  265.                 }
  266.             } catch (AuthorizationException ae) {
  267.             }
  268.         } finally {
  269.             pm.setAuthenticatedUserURI(currentUser);
  270.         }
  271.     }

  273.     private void updateEntries(Set<Entry> entries) {
  274.         URI currentUser = pm.getAuthenticatedUserURI();
  275.         try {
  276.             pm.setAuthenticatedUserURI(pm.getGuestUser().getURI());
  277.             synchronized (repository) {
  278.                 RepositoryConnection rc = null;
  279.                 try {
  280.                     rc = repository.getConnection();
  281.                     rc.begin();
  282.                     for (Entry e : entries) {
  283.                         updateEntry(e, rc);
  284.                     }
  285.                     rc.commit();
  286.                 } catch (RepositoryException re) {
  287.                     try {
  288.                         if (rc != null) {
  289.                             rc.rollback();
  290.                         }
  291.                     } catch (RepositoryException re1) {
  292.                         log.error(re1.getMessage());
  293.                     }
  294.                     log.error(re.getMessage());
  295.                 } finally {
  296.                     if (rc != null) {
  297.                         try {
  298.                             rc.close();
  299.                         } catch (RepositoryException re2) {
  300.                             log.error(re2.getMessage());
  301.                         }
  302.                     }
  303.                 }
  304.             }
  305.         } finally {
  306.             pm.setAuthenticatedUserURI(currentUser);
  307.         }
  308.     }

  310.     private void updateEntry(Entry e, RepositoryConnection rc) throws RepositoryException {
  311.         if (e == null) {
  312.             return;
  313.         }
  314.         
  315.         // If entry is ResourceType.Context we update all its
  316.         // entries, just in case the ACL has changed
  317.         if (GraphType.Context.equals(e.getGraphType()) && EntryType.Local.equals(e.getEntryType())) {
  318.             String contextURI = e.getResourceURI().toString();
  319.             String id = contextURI.substring(contextURI.lastIndexOf("/") + 1);
  320.             Context context = rm.getContextManager().getContext(id);
  321.             if (context != null) {
  322.                 Set<URI> entries = context.getEntries();
  323.                 for (URI entryURI : entries) {
  324.                     if (entryURI != null) {
  325.                         try {
  326.                             updateEntry(rm.getContextManager().getEntry(entryURI), rc);
  327.                         } catch (AuthorizationException ae) {
  328.                             continue;
  329.                         }
  330.                     }
  331.                 }
  332.             }
  333.         } else {
  334.             log.debug("Processing entry: " + e.getEntryURI());
  335.             removeEntry(e, rc);
  336.             addEntry(e, rc);
  337.         }
  338.     }

  340.     private void removeEntries(Set<Entry> entries) {
  341.         URI currentUser = pm.getAuthenticatedUserURI();
  342.         try {
  343.             pm.setAuthenticatedUserURI(pm.getGuestUser().getURI());
  344.             synchronized (repository) {
  345.                 RepositoryConnection rc = null;
  346.                 try {
  347.                     rc = repository.getConnection();
  348.                     rc.begin();
  349.                     for (Entry e : entries) {
  350.                         removeEntry(e, rc);
  351.                     }
  352.                     rc.commit();
  353.                 } catch (RepositoryException re) {
  354.                     try {
  355.                         rc.rollback();
  356.                     } catch (RepositoryException re1) {
  357.                         log.error(re1.getMessage());
  358.                     }
  359.                     log.error(re.getMessage());
  360.                 } finally {
  361.                     if (rc != null) {
  362.                         try {
  363.                             rc.close();
  364.                         } catch (RepositoryException re2) {
  365.                             log.error(re2.getMessage());
  366.                         }
  367.                     }
  368.                 }
  369.             }
  370.         } finally {
  371.             pm.setAuthenticatedUserURI(currentUser);
  372.         }
  373.     }

  375.     private void removeEntry(Entry e, RepositoryConnection rc) throws RepositoryException {
  376.         PrincipalManager pm = e.getRepositoryManager().getPrincipalManager();
  377.         URI currentUser = pm.getAuthenticatedUserURI();
  378.         try {
  379.             // we need to be admin, in case the ACL has become
  380.             // more restrictive since adding the entry
  381.             pm.setAuthenticatedUserURI(pm.getAdminUser().getURI());
  382.             ValueFactory vf = repository.getValueFactory();
  383.             IRI contextURI = vf.createIRI(e.getContext().getURI().toString());

  385.             IRI entryNG = vf.createIRI(e.getEntryURI().toString());
  386.             IRI mdNG = vf.createIRI(e.getLocalMetadataURI().toString());
  387.             IRI resNG = vf.createIRI(e.getResourceURI().toString());
  388.             IRI extMdNG = null;

  390.             if (e.getExternalMetadataURI() != null) {
  391.                 extMdNG = vf.createIRI(e.getCachedExternalMetadataURI().toString());
  392.             }

  394.             if (extMdNG != null) {
  395.                 rc.remove(rc.getStatements((Resource) null, (IRI) null, (Value) null, false, entryNG, mdNG, extMdNG, resNG), contextURI, entryNG, mdNG, extMdNG, resNG);
  396.             } else {
  397.                 rc.remove(rc.getStatements((Resource) null, (IRI) null, (Value) null, false, entryNG, mdNG, extMdNG, resNG), contextURI, entryNG, mdNG, resNG);
  398.             }
  399.         } finally {
  400.             pm.setAuthenticatedUserURI(currentUser);
  401.         }
  402.     }

  404.     public void rebuildRepository() {
  405.         synchronized (repository) {
  406.             if (rebuilding) {
  407.                 log.warn("The public repository is already being rebuilt: ignoring additional rebuilding requests");
  408.                 return;
  409.             } else {
  410.                 rebuilding = true;
  411.             }
  412.         }

  414.         log.info("Rebuilding public repository");

  416.         synchronized (repository) {
  417.             RepositoryConnection rc = null;
  418.             try {
  419.                 rc = repository.getConnection();
  420.                 Date before = new Date();
  421.                 rc.begin();
  422.                 rc.clear();
  423.                 log.info("Clearing public repository took " + (new Date().getTime() - before.getTime()) + " ms");

  425.                 ContextManager cm = rm.getContextManager();
  426.                 Set<URI> contexts = cm.getEntries();

  428.                 for (URI contextURI : contexts) {
  429.                     String id = contextURI.toString().substring(contextURI.toString().lastIndexOf("/") + 1);
  430.                     Context context = cm.getContext(id);
  431.                     if (context != null) {
  432.                         log.info("Adding context " + contextURI + " to public repository");
  433.                         before = new Date();
  434.                         Set<URI> entries = context.getEntries();
  435.                         log.info("Fetching entries took " + (new Date().getTime() - before.getTime()) + " ms");
  436.                         before = new Date();
  437.                         Date timeTracker = new Date();
  438.                         long publicEntryCount = 0;
  439.                         long processedCount = 0;
  440.                         for (URI entryURI : entries) {
  441.                             if (entryURI != null) {
  442.                                 processedCount++;
  443.                                 if ((new Date().getTime() - timeTracker.getTime()) > 60000) {
  444.                                     if (processedCount > 0) {
  445.                                         log.debug("Average time per entry after " + (new Date().getTime() - before.getTime()) + " ms: " + ((new Date().getTime() - before.getTime()) / processedCount) + " ms");
  446.                                         timeTracker = new Date();
  447.                                     }
  448.                                 }
  449.                                 try {
  450.                                     Entry entry = cm.getEntry(entryURI);
  451.                                     if (entry == null) {
  452.                                         continue;
  453.                                     }
  454.                                     addEntry(entry, rc);
  455.                                     publicEntryCount++;
  456.                                 } catch (AuthorizationException ae) {
  457.                                     continue;
  458.                                 }
  459.                             }
  460.                         }
  461.                         log.info("Added " + publicEntryCount + " entries to public repository");
  462.                         log.info("Total time for context: " + (new Date().getTime() - before.getTime()) + " ms");
  463.                         if (entries.size() > 0) {
  464.                             log.debug("Total average time per entry: " + ((new Date().getTime() - before.getTime()) / entries.size()) + " ms");
  465.                         }
  466.                         log.info("Done processing context " + contextURI);
  467.                     }
  468.                 }
  469.                 rc.commit();
  470.             } catch (RepositoryException re) {
  471.                 try {
  472.                     rc.rollback();
  473.                 } catch (RepositoryException re1) {
  474.                     log.error(re1.getMessage());
  475.                 }
  476.                 log.error(re.getMessage());
  477.             } finally {
  478.                 try {
  479.                     rc.close();
  480.                 } catch (RepositoryException re) {
  481.                     log.error(re.getMessage());
  482.                 }
  483.                 log.info("Rebuild of public repository complete");
  484.                 log.info("Number of triples in public repository: " + getTripleCount());
  485.                 rebuilding = false;
  486.             }
  487.         }
  488.     }

  490.     private boolean isAdministrative(Entry e) {
  491.         GraphType gt = e.getGraphType();
  492.         if (GraphType.Graph.equals(gt) ||
  493.                 GraphType.String.equals(gt) ||
  494.                 GraphType.None.equals(gt) ||
  495.                 GraphType.List.equals(gt)) {
  496.             return false;
  497.         }
  498.         return true;
  499.     }

  501.     public long getTripleCount() {
  502.         long amountTriples = 0;
  503.         RepositoryConnection rc = null;
  504.         try {
  505.             rc = repository.getConnection();
  506.             amountTriples = rc.size();
  507.         } catch (RepositoryException re) {
  508.             log.error(re.getMessage());
  509.         } finally {
  510.             if (rc != null) {
  511.                 try {
  512.                     rc.close();
  513.                 } catch (RepositoryException e) {
  514.                     log.error(e.getMessage());
  515.                 }
  516.             }
  517.         }
  518.         return amountTriples;
  519.     }

  521.     public void shutdown() {
  522.         if (entrySubmitter != null) {
  523.             entrySubmitter.interrupt();
  524.         }

  526.         try {
  527.             repository.shutDown();
  528.         } catch (RepositoryException e) {
  529.             log.error("Error when shutting down public repository: " + e.getMessage());
  530.         }
  531.     }

  533. }
Created with JaCoCo 0.8.12.202403310830