DocumentNodeStore implements NodeStore, RevisionContext, Observable, Clusterable, NodeStateDiffer { private BackgroundReadStats backgroundRead() { return new ExternalChange(this) { @Override void invalidateCache(@NotNull Iterable<String> paths) { stats.cacheStats = store.invalidateCache(pathToId(paths)); } @Override void invalidateCache() { stats.cacheStats = store.invalidateCache(); } @Override void updateHead(@NotNull Set<Revision> externalChanges, @NotNull RevisionVector sweepRevs, @Nullable Iterable<String> changedPaths) { long time = clock.getTime(); backgroundOperationLock.writeLock().lock(); try { stats.lock = clock.getTime() - time; RevisionVector oldHead = getHeadRevision(); RevisionVector newHead = oldHead; for (Revision r : externalChanges) { newHead = newHead.update(r); } setRoot(newHead); sweepRevisions = sweepRevisions.pmax(sweepRevs); commitQueue.headRevisionChanged(); time = clock.getTime(); if (changedPaths != null) { try { JournalEntry.applyTo(changedPaths, diffCache, ROOT, oldHead, newHead); } catch (Exception e1) { LOG.error("backgroundRead: Exception while processing external changes from journal: " + e1, e1); } } stats.populateDiffCache = clock.getTime() - time; time = clock.getTime(); ChangeSet changeSet = getChangeSetBuilder().build(); LOG.debug("Dispatching external change with ChangeSet {}", changeSet); dispatcher.contentChanged(getRoot().fromExternalChange(), newCommitInfo(changeSet, getJournalPropertyHandler())); } finally { backgroundOperationLock.writeLock().unlock(); } stats.dispatchChanges = clock.getTime() - time; } }.process(); } DocumentNodeStore(DocumentNodeStoreBuilder<?> builder); void dispose(); @NotNull DocumentStore getDocumentStore(); void setAsyncDelay(int delay); int getAsyncDelay(); void setMaxBackOffMillis(int time); int getMaxBackOffMillis(); int getChangeSetMaxItems(); void setChangeSetMaxItems(int changeSetMaxItems); int getChangeSetMaxDepth(); void setChangeSetMaxDepth(int changeSetMaxDepth); @NotNull ClusterNodeInfo getClusterInfo(); CacheStats getNodeCacheStats(); CacheStats getNodeChildrenCacheStats(); @NotNull Iterable<CacheStats> getDiffCacheStats(); Cache<PathRev, DocumentNodeState> getNodeCache(); Cache<NamePathRev, DocumentNodeState.Children> getNodeChildrenCache(); Predicate<Path> getNodeCachePredicate(); int getPendingWriteCount(); boolean isDisableBranches(); @NotNull PropertyState createPropertyState(String name, String value); @Nullable DocumentNodeState getNode(@NotNull final Path path,
                                     @NotNull final RevisionVector rev); BundlingConfigHandler getBundlingConfigHandler(); @Override boolean compare(@NotNull final AbstractDocumentNodeState node,
                           @NotNull final AbstractDocumentNodeState base,
                           @NotNull NodeStateDiff diff); @Override Closeable addObserver(Observer observer); @NotNull @Override DocumentNodeState getRoot(); @NotNull @Override NodeState merge(@NotNull NodeBuilder builder,
                           @NotNull CommitHook commitHook,
                           @NotNull CommitInfo info); @NotNull @Override NodeState rebase(@NotNull NodeBuilder builder); @Override NodeState reset(@NotNull NodeBuilder builder); @Override @NotNull BlobStoreBlob createBlob(InputStream inputStream); @Override Blob getBlob(@NotNull String reference); Blob getBlobFromBlobId(String blobId); @NotNull @Override String checkpoint(long lifetime, @NotNull Map<String, String> properties); @NotNull @Override String checkpoint(long lifetime); @NotNull @Override Map<String, String> checkpointInfo(@NotNull String checkpoint); @NotNull @Override Iterable<String> checkpoints(); @Nullable @Override NodeState retrieve(@NotNull String checkpoint); @Override boolean release(@NotNull String checkpoint); @Override UnmergedBranches getBranches(); @Override UnsavedModifications getPendingModifications(); @Override int getClusterId(); @Override @NotNull RevisionVector getHeadRevision(); @Override @NotNull Revision newRevision(); @Override @NotNull Clock getClock(); @Override String getCommitValue(@NotNull Revision changeRevision,
                                 @NotNull NodeDocument doc); void runBackgroundOperations(); @Nullable MarkSweepGarbageCollector createBlobGarbageCollector(long blobGcMaxAgeInSecs, String repositoryId,
        Whiteboard whiteboard, StatisticsProvider statisticsProvider); DocumentNodeStoreMBean getMBean(); BlobStore getBlobStore(); Iterator<ReferencedBlob> getReferencedBlobsIterator(); DiffCache getDiffCache(); Checkpoints getCheckpoints(); @NotNull VersionGarbageCollector getVersionGarbageCollector(); @NotNull JournalGarbageCollector getJournalGarbageCollector(); @NotNull LastRevRecoveryAgent getLastRevRecoveryAgent(); @Override String getInstanceId(); @Override String getVisibilityToken(); @Override boolean isVisible(@NotNull String visibilityToken, long maxWaitMillis); DocumentNodeStoreStatsCollector getStatsCollector(); DocumentNodeStateCache getNodeStateCache(); void setNodeStateCache(DocumentNodeStateCache nodeStateCache); JournalPropertyHandlerFactory getJournalPropertyHandlerFactory(); static final FormatVersion VERSION; static final List<String> META_PROP_NAMES; static final String SYS_PROP_DISABLE_JOURNAL; }
@Test public void backgroundRead() throws Exception { final Semaphore semaphore = new Semaphore(1); DocumentStore docStore = new MemoryDocumentStore(); DocumentStore testStore = new TimingDocumentStoreWrapper(docStore) { @Override public CacheInvalidationStats invalidateCache(Iterable<String> keys) { super.invalidateCache(keys); semaphore.acquireUninterruptibly(); semaphore.release(); return null; } }; final DocumentNodeStore store1 = builderProvider.newBuilder().setAsyncDelay(0) .setDocumentStore(testStore).setClusterId(1).getNodeStore(); DocumentNodeStore store2 = builderProvider.newBuilder().setAsyncDelay(0) .setDocumentStore(docStore).setClusterId(2).getNodeStore(); NodeBuilder builder = store2.getRoot().builder(); builder.child("node2"); store2.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY); store2.runBackgroundOperations(); assertFalse(store1.getRoot().hasChildNode("node2")); builder = store1.getRoot().builder(); builder.child("node1"); NodeState root = store1.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY); semaphore.acquireUninterruptibly(); Thread t = new Thread(new Runnable() { @Override public void run() { store1.runBackgroundOperations(); } }); t.start(); while (!semaphore.hasQueuedThreads()) { Thread.sleep(10); } try { assertFalse(root.hasChildNode("node2")); } finally { semaphore.release(); } t.join(); root = store1.getRoot(); assertTrue(root.hasChildNode("node2")); }