import org.apache.commons.rdf.api.Dataset; //导入依赖的package包/类
@Test
public void testEntity() {
    final Triple triple = rdf.createTriple(identifier, RDFS.label, rdf.createLiteral("A label"));

    when(mockResource.stream(eq(Trellis.PreferUserManaged))).thenAnswer(x -> of(triple));
    when(mockLdpRequest.getPath()).thenReturn("resource");

    final PatchHandler patchHandler = new PatchHandler(mockLdpRequest, insert,
            mockResourceService, mockIoService, null);

    final Response res = patchHandler.updateResource(mockResource).build();
    assertEquals(NO_CONTENT, res.getStatusInfo());

    verify(mockIoService).update(any(Graph.class), eq(insert), eq(identifier.getIRIString()));

    verify(mockResourceService).put(eq(identifier), eq(LDP.RDFSource), any(Dataset.class));
}
 

import org.apache.commons.rdf.api.Dataset; //导入依赖的package包/类
@BeforeEach
public void setUp() {
    initMocks(this);
    when(mockBinaryService.getIdentifierSupplier()).thenReturn(() -> "file:" + randomUUID());
    when(mockResourceService.put(any(IRI.class), any(IRI.class), any(Dataset.class)))
        .thenReturn(completedFuture(true));
    when(mockResourceService.skolemize(any(Literal.class))).then(returnsFirstArg());
    when(mockResourceService.skolemize(any(IRI.class))).then(returnsFirstArg());
    when(mockResourceService.skolemize(any(BlankNode.class))).thenAnswer(inv ->
            rdf.createIRI(TRELLIS_BNODE_PREFIX + ((BlankNode) inv.getArgument(0)).uniqueReference()));

    when(mockRequest.getSession()).thenReturn(new HttpSession());
    when(mockRequest.getPath()).thenReturn("");
    when(mockRequest.getBaseUrl()).thenReturn(baseUrl);
    when(mockResourceService.toInternal(any(RDFTerm.class), any())).thenAnswer(inv -> {
        final RDFTerm term = (RDFTerm) inv.getArgument(0);
        if (term instanceof IRI) {
            final String iri = ((IRI) term).getIRIString();
            if (iri.startsWith(baseUrl)) {
                return rdf.createIRI(TRELLIS_PREFIX + iri.substring(baseUrl.length()));
            }
        }
        return term;
    });
}
 

import org.apache.commons.rdf.api.Dataset; //导入依赖的package包/类
@Test
public void testEntity2() throws IOException {
    final IRI identifier = rdf.createIRI("trellis:newresource");
    final File entity = new File(getClass().getResource("/simpleData.txt").getFile());
    when(mockRequest.getContentType()).thenReturn("text/plain");

    final PostHandler postHandler = new PostHandler(mockRequest, "newresource", entity,
            mockResourceService, mockIoService, mockBinaryService, null);

    final Response res = postHandler.createResource().build();
    assertEquals(CREATED, res.getStatusInfo());
    assertTrue(res.getLinks().stream().anyMatch(hasType(LDP.Resource)));
    assertFalse(res.getLinks().stream().anyMatch(hasType(LDP.RDFSource)));
    assertFalse(res.getLinks().stream().anyMatch(hasType(LDP.Container)));
    assertTrue(res.getLinks().stream().anyMatch(hasType(LDP.NonRDFSource)));
    assertEquals(create(baseUrl + "newresource"), res.getLocation());

    verify(mockIoService, never()).read(any(), any(), any());

    verify(mockBinaryService).setContent(iriArgument.capture(), any(InputStream.class),
            metadataArgument.capture());
    assertTrue(iriArgument.getValue().getIRIString().startsWith("file:"));
    assertEquals("text/plain", metadataArgument.getValue().get(CONTENT_TYPE));

    verify(mockResourceService).put(eq(identifier), eq(LDP.NonRDFSource), any(Dataset.class));
}
 

import org.apache.commons.rdf.api.Dataset; //导入依赖的package包/类
/**
 * Process the element
 * @param c the context
 */
@ProcessElement
public void processElement(final ProcessContext c) {
    final KV<String, String> element = c.element();
    final File dir = resourceDirectory(config, element.getKey());
    if (!isNull(dir)) {
        LOGGER.debug("Writing {} to directory: {}", graph, dir);
        try (final Dataset dataset = deserialize(element.getValue())) {
            if (VersionedResource.write(dir,
                        add ? empty() : dataset.stream(of(rdf.createIRI(graph)), null, null, null),
                        add ? dataset.stream(of(rdf.createIRI(graph)), null, null, null) : empty(), now())) {
                c.output(c.element());
            } else {
                logError(add, graph, element.getKey());
            }
        } catch (final Exception ex) {
            LOGGER.error("Error processing graph: {}", ex.getMessage());
        }
    } else {
        LOGGER.error("Unable to write {} quads to {}", graph, element.getKey());
    }
}
 

import org.apache.commons.rdf.api.Dataset; //导入依赖的package包/类
/**
 * Process the element
 * @param c the context
 */
@ProcessElement
public void processElement(final ProcessContext c) {
    final KV<String, String> element = c.element();
    final Dataset data = deserialize(element.getValue());
    final String baseUrl = baseUrls.get(element.getKey().split(":", 2)[1].split("/")[0]);
    if (nonNull(baseUrl) && nonNull(data)) {
        final String identifier = toExternalTerm(element.getKey(), baseUrl);
        final Notification notification = new Notification(identifier, data);
        LOGGER.debug("Serializing notification for {}", element.getKey());
        LOGGER.debug("Using baseURL: {} for {}", baseUrl, element.getKey());
        service.serialize(notification).ifPresent(evt -> c.output(of(element.getKey(), evt)));
    } else {
        LOGGER.warn("Unable to serialize notification: baseUrl or data values not present for: {}",
                element.getKey());
    }
}
 

import org.apache.commons.rdf.api.Dataset; //导入依赖的package包/类
/**
 * Read the state of the resource data at a particular point in time
 * @param directory the directory
 * @param identifier the identifier
 * @param time the time
 * @return the resource data, if it exists
 */
public static Optional<ResourceData> read(final File directory, final IRI identifier, final Instant time) {
    LOGGER.debug("Reading journal to generate the resource data");
    return of(new File(directory, RESOURCE_JOURNAL)).filter(File::exists).flatMap(file -> {
        final List<Instant> mementos = new ArrayList<>();
        final List<VersionRange> ranges = asTimeMap(file);
        ranges.stream().map(VersionRange::getFrom).findFirst().ifPresent(mementos::add);
        ranges.stream().map(VersionRange::getUntil).forEachOrdered(mementos::add);

        try (final Stream<Quad> stream = asStream(rdf, file, identifier, time)) {
            try (final Dataset dataset = stream.filter(isResourceTriple).collect(toDataset())) {
                LOGGER.debug("Creating resource: {} at {}", identifier, time);
                return from(identifier, dataset, mementos);
            } catch (final Exception ex) {
                throw new RuntimeRepositoryException("Error processing dataset", ex);
            }
        }
    });
}
 

import org.apache.commons.rdf.api.Dataset; //导入依赖的package包/类
@Test
public void testWriteResource() {
    final Dataset data = rdf.createDataset();
    data.add(rdf.createQuad(Trellis.PreferUserManaged, testResource, DC.title, rdf.createLiteral("A title")));
    data.add(rdf.createQuad(Trellis.PreferServerManaged, testResource, type, LDP.RDFSource));
    assertFalse(service.get(testResource).isPresent());
    assertFalse(service.get(testResource, now()).isPresent());

    assertTrue(service.put(testResource, data));
    final Optional<Resource> res = service.get(testResource, now());
    assertTrue(res.isPresent());
    res.ifPresent(r -> {
        assertEquals(LDP.RDFSource, r.getInteractionModel());
        assertEquals(testResource, r.getIdentifier());
        assertTrue(r.stream().anyMatch(q -> q.getPredicate().equals(DC.title)));
        assertTrue(r.getModified().isBefore(now()));
    });
    final Optional<Resource> res2 = service.get(testResource);
    assertTrue(res2.isPresent());
    res2.ifPresent(r -> {
        assertEquals(LDP.RDFSource, r.getInteractionModel());
        assertEquals(testResource, r.getIdentifier());
        assertTrue(r.stream().anyMatch(q -> q.getPredicate().equals(DC.title)));
        assertTrue(r.getModified().isBefore(now()));
    });
}
 

import org.apache.commons.rdf.api.Dataset; //导入依赖的package包/类
/**
 * Process the element
 * @param c the context
 */
@ProcessElement
public void processElement(final ProcessContext c) {
    final KV<String, String> element = c.element();
    final File dir = resourceDirectory(dataLocation, element.getKey());
    if (!isNull(dir)) {
        LOGGER.debug("Writing {} to directory: {}", graph, dir);
        try (final Dataset dataset = deserialize(element.getValue())) {
            if (VersionedResource.write(dir,
                        add ? empty() : dataset.stream(of(rdf.createIRI(graph)), null, null, null),
                        add ? dataset.stream(of(rdf.createIRI(graph)), null, null, null) : empty(), now())) {
                c.output(c.element());
            } else {
                logError(add, graph, element.getKey());
            }
        } catch (final Exception ex) {
            LOGGER.error("Error processing graph: {}", ex.getMessage());
        }
    } else {
        LOGGER.error("Unable to write {} quads to {}", graph, element.getKey());
    }
}
 

import org.apache.commons.rdf.api.Dataset; //导入依赖的package包/类
/**
 * Process the element
 * @param c the context
 */
@ProcessElement
public void processElement(final ProcessContext c) {
    final KV<String, String> element = c.element();
    final Dataset data = deserialize(element.getValue());
    if (nonNull(baseUrl) && nonNull(data)) {
        final String identifier = toExternalTerm(element.getKey());
        final Notification notification = new Notification(identifier, data);
        LOGGER.debug("Serializing notification for {}", element.getKey());
        LOGGER.debug("Using baseURL: {} for {}", baseUrl, element.getKey());
        service.serialize(notification).ifPresent(evt -> c.output(of(element.getKey(), evt)));
    } else {
        LOGGER.warn("Unable to serialize notification: baseUrl or data values not present for: {}",
                element.getKey());
    }
}
 

import org.apache.commons.rdf.api.Dataset; //导入依赖的package包/类
private Optional<ProducerRecord<String, String>> buildMembershipMessage(final String topic, final IRI resource,
        final Resource parent, final Dataset dataset) throws Exception {
    try (final Dataset data = rdf.createDataset()) {
        if (DirectContainer.equals(parent.getInteractionModel())) {
            parent.getMembershipResource().ifPresent(member -> {
                parent.getMemberRelation().ifPresent(relation ->
                    data.add(rdf.createQuad(PreferMembership, member, relation, resource)));
                parent.getMemberOfRelation().ifPresent(relation ->
                    data.add(rdf.createQuad(PreferMembership, resource, relation, member)));
            });
        } else if (IndirectContainer.equals(parent.getInteractionModel())) {
            parent.getMembershipResource().ifPresent(member ->
                parent.getMemberRelation().ifPresent(relation ->
                    parent.getInsertedContentRelation().ifPresent(inserted ->
                        dataset.stream(of(PreferUserManaged), null, inserted, null).sequential().forEachOrdered(q ->
                            data.add(rdf.createQuad(PreferMembership, member, relation, q.getObject()))))));
        }
        final Optional<String> key = data.stream(of(PreferMembership), null, null, null).map(Quad::getSubject)
            .filter(x -> x instanceof IRI).map(x -> (IRI) x).map(IRI::getIRIString).findFirst();
        if (key.isPresent()) {
            dataset.stream(of(PreferAudit), null, null, null).map(auditTypeMapper).forEachOrdered(data::add);
            return of(new ProducerRecord<>(topic, key.get(), serialize(data)));
        }
        return empty();
    }
}
 

import org.apache.commons.rdf.api.Dataset; //导入依赖的package包/类
private Consumer<Resource> emitToParent(final IRI identifier, final Dataset dataset,
        final List<Future<RecordMetadata>> results) {
    final Boolean isCreate = dataset.contains(of(PreferAudit), null, type, Create);
    final Boolean isDelete = dataset.contains(of(PreferAudit), null, type, Delete);
    final String containmentTopic = isDelete ? TOPIC_LDP_CONTAINMENT_DELETE : TOPIC_LDP_CONTAINMENT_ADD;
    final String membershipTopic = isDelete ? TOPIC_LDP_MEMBERSHIP_DELETE : TOPIC_LDP_MEMBERSHIP_ADD;

    return container -> {
        if (isDelete || isCreate) {
            try {
                LOGGER.info("Sending to parent: {}", container.getIdentifier());
                results.add(producer.send(buildContainmentMessage(containmentTopic, identifier, container,
                                dataset)));

                buildMembershipMessage(membershipTopic, identifier, container, dataset).ifPresent(msg -> {
                        LOGGER.info("Sending to member resource: {}", container.getMembershipResource());
                        results.add(producer.send(msg));
                });
            } catch (final Exception ex) {
                LOGGER.error("Error processing dataset: {}", ex.getMessage());
            }
        }
    };
}
 

import org.apache.commons.rdf.api.Dataset; //导入依赖的package包/类
/**
 * Read the state of the resource data at a particular point in time
 * @param directory the directory
 * @param identifier the identifier
 * @param time the time
 * @return the resource data, if it exists
 */
public static Optional<ResourceData> read(final File directory, final IRI identifier, final Instant time) {
    LOGGER.debug("Reading journal to generate the resource data");
    return of(new File(directory, RESOURCE_JOURNAL)).filter(File::exists).flatMap(file -> {
        final List<Instant> mementos = new ArrayList<>();
        final List<VersionRange> ranges = asTimeMap(file);
        ranges.stream().map(VersionRange::getFrom).findFirst().ifPresent(mementos::add);
        ranges.stream().map(VersionRange::getUntil).forEachOrdered(mementos::add);

        try (final Stream<Quad> stream = asStream(rdf, file, identifier, time)) {
            try (final Dataset dataset = stream.filter(isResourceTriple).collect(toDataset())) {
                LOGGER.debug("Creating resource: {} at {}", identifier, time);
                return from(identifier, dataset, mementos);
            } catch (final Exception ex) {
                throw new RuntimeTrellisException("Error processing dataset", ex);
            }
        }
    });
}
 

import org.apache.commons.rdf.api.Dataset; //导入依赖的package包/类
@Test
public void testWriteResource() throws Exception {
    final Dataset data = rdf.createDataset();
    data.add(rdf.createQuad(Trellis.PreferUserManaged, testResource, DC.title, rdf.createLiteral("A title")));
    data.add(rdf.createQuad(Trellis.PreferServerManaged, testResource, type, LDP.RDFSource));
    assertFalse(service.get(testResource).isPresent());
    assertFalse(service.get(testResource, now()).isPresent());

    assertTrue(service.put(testResource, LDP.RDFSource, data).get());
    final Optional<Resource> res = service.get(testResource, now());
    assertTrue(res.isPresent());
    res.ifPresent(r -> {
        assertEquals(LDP.RDFSource, r.getInteractionModel());
        assertEquals(testResource, r.getIdentifier());
        assertTrue(r.stream().anyMatch(q -> q.getPredicate().equals(DC.title)));
        assertTrue(r.getModified().isBefore(now()));
    });
    final Optional<Resource> res2 = service.get(testResource);
    assertTrue(res2.isPresent());
    res2.ifPresent(r -> {
        assertEquals(LDP.RDFSource, r.getInteractionModel());
        assertEquals(testResource, r.getIdentifier());
        assertTrue(r.stream().anyMatch(q -> q.getPredicate().equals(DC.title)));
        assertTrue(r.getModified().isBefore(now()));
    });
}
 

import org.apache.commons.rdf.api.Dataset; //导入依赖的package包/类
/**
 * Collect a stream of Quads into a Dataset.
 *
 * @return a dataset
 */
public static Collector<Quad, ?, Dataset> toDataset() {
    return of(rdf::createDataset, Dataset::add, (left, right) -> {
        right.iterate().forEach(left::add);
        return left;
    }, UNORDERED);
}
 

import org.apache.commons.rdf.api.Dataset; //导入依赖的package包/类
@Test
public void testCollectDataset() {
    final Dataset dataset = generate(() -> rdf.createQuad(getIRI(), getIRI(), getIRI(), getIRI()))
        .parallel().limit(size).collect(toDataset());

    assertTrue(size >= dataset.size());
}
 

import org.apache.commons.rdf.api.Dataset; //导入依赖的package包/类
@Test
public void testAuditCreation() {
    final Dataset dataset = rdf.createDataset();
    final AuditService svc = new DefaultAuditService();
    svc.creation(subject, mockSession).forEach(dataset::add);
    assertTrue(dataset.getGraph(Trellis.PreferAudit).filter(graph -> graph.size() == dataset.size()).isPresent());
    assertTrue(dataset.contains(null, null, type, PROV.Activity));
    assertTrue(dataset.contains(null, null, type, AS.Create));
    assertTrue(dataset.contains(null, subject, PROV.wasGeneratedBy, null));
    assertTrue(dataset.contains(null, null, PROV.wasAssociatedWith, Trellis.AnonymousAgent));
    assertTrue(dataset.contains(null, null, PROV.actedOnBehalfOf, Trellis.AdministratorAgent));
    assertTrue(dataset.contains(null, null, PROV.startedAtTime,
                rdf.createLiteral(created.toString(), XSD.dateTime)));
    assertEquals(6L, dataset.size());
}
 

import org.apache.commons.rdf.api.Dataset; //导入依赖的package包/类
@Test
public void testAuditDeletion() {
    final Dataset dataset = rdf.createDataset();
    final AuditService svc = new DefaultAuditService();
    svc.deletion(subject, mockSession).forEach(dataset::add);
    assertTrue(dataset.getGraph(Trellis.PreferAudit).filter(graph -> graph.size() == dataset.size()).isPresent());
    assertTrue(dataset.contains(null, null, type, PROV.Activity));
    assertTrue(dataset.contains(null, null, type, AS.Delete));
    assertTrue(dataset.contains(null, subject, PROV.wasGeneratedBy, null));
    assertTrue(dataset.contains(null, null, PROV.wasAssociatedWith, Trellis.AnonymousAgent));
    assertTrue(dataset.contains(null, null, PROV.actedOnBehalfOf, Trellis.AdministratorAgent));
    assertTrue(dataset.contains(null, null, PROV.startedAtTime,
                rdf.createLiteral(created.toString(), XSD.dateTime)));
    assertEquals(6L, dataset.size());
}
 

import org.apache.commons.rdf.api.Dataset; //导入依赖的package包/类
@Test
public void testAuditUpdate() {
    final Dataset dataset = rdf.createDataset();
    final AuditService svc = new DefaultAuditService();
    svc.update(subject, mockSession).forEach(dataset::add);
    assertTrue(dataset.getGraph(Trellis.PreferAudit).filter(graph -> graph.size() == dataset.size()).isPresent());
    assertTrue(dataset.contains(null, null, type, PROV.Activity));
    assertTrue(dataset.contains(null, null, type, AS.Update));
    assertTrue(dataset.contains(null, subject, PROV.wasGeneratedBy, null));
    assertTrue(dataset.contains(null, null, PROV.wasAssociatedWith, Trellis.AnonymousAgent));
    assertTrue(dataset.contains(null, null, PROV.actedOnBehalfOf, Trellis.AdministratorAgent));
    assertTrue(dataset.contains(null, null, PROV.startedAtTime,
                rdf.createLiteral(created.toString(), XSD.dateTime)));
    assertEquals(6L, dataset.size());
}
 

import org.apache.commons.rdf.api.Dataset; //导入依赖的package包/类
@Test
public void testPostInterrupted() throws Exception {
    when(mockResource.getInteractionModel()).thenReturn(LDP.Container);
    when(mockResourceService.put(eq(rdf.createIRI(TRELLIS_PREFIX + CHILD_PATH)), eq(LDP.RDFSource),
                any(Dataset.class))).thenReturn(mockFuture);
    doThrow(new InterruptedException("Expected InterruptedException")).when(mockFuture).get();

    final Response res = target(RESOURCE_PATH).request().header("Slug", "child")
        .post(entity("<> <http://purl.org/dc/terms/title> \"A title\" .", TEXT_TURTLE_TYPE));

    assertEquals(INTERNAL_SERVER_ERROR, res.getStatusInfo());
}
 

import org.apache.commons.rdf.api.Dataset; //导入依赖的package包/类
@Test
public void testPostFutureException() throws Exception {
    when(mockResource.getInteractionModel()).thenReturn(LDP.Container);
    when(mockResourceService.put(eq(rdf.createIRI(TRELLIS_PREFIX + CHILD_PATH)), eq(LDP.RDFSource),
                any(Dataset.class))).thenReturn(mockFuture);
    doThrow(ExecutionException.class).when(mockFuture).get();

    final Response res = target(RESOURCE_PATH).request().header("Slug", "child")
        .post(entity("<> <http://purl.org/dc/terms/title> \"A title\" .", TEXT_TURTLE_TYPE));

    assertEquals(INTERNAL_SERVER_ERROR, res.getStatusInfo());
}