Java 类org.apache.camel.impl.DefaultUnitOfWork 实例源码

/**
 * Creates a new {@link Exchange} from <code>packet</code> using the
 * current {@link CamelContext}.
 * 
 * @param packet message packet.
 * @return a new message exchange.
 */
protected Exchange createExchange(PlatformPacket packet) {
    DefaultExchange exchange = new DefaultExchange(camelContext);

    exchange.setUnitOfWork(new DefaultUnitOfWork(exchange));
    exchange.setProperties(new HashMap<>(packet.getExchangeProperties()));
    exchange.getIn().setHeaders(new HashMap<>(packet.getMessageProperties()));

    setInBody(packet.getMessageBody(), exchange);

    return exchange;
}

protected void setUp() throws Exception {
    super.setUp();

    deleteDirectory("target/cachedir");
    createDirectory("target/cachedir");

    exchange = new DefaultExchange(context);
    UnitOfWork uow = new DefaultUnitOfWork(exchange);
    exchange.setUnitOfWork(uow);
}

@Test
public void testTransactedGetProducer() throws Exception {
    final String baseUrl = "http://localhost:8080/rest";
    final String path = "/transact";
    final String path2 = "/transact2";
    final String tx = "tx:12345";
    final URI uri = create(baseUrl + "/" + tx + path);
    final URI uri2 = create(baseUrl + "/" + tx + path2);
    final URI commitUri = URI.create(baseUrl + "/" + tx + FcrepoConstants.COMMIT);
    final URI beginUri = URI.create(baseUrl + FcrepoConstants.TRANSACTION);
    final int status = 200;
    final ByteArrayInputStream body = new ByteArrayInputStream(TestUtils.rdfXml.getBytes());
    final ByteArrayInputStream body2 = new ByteArrayInputStream(TestUtils.rdfTriples.getBytes());
    final DefaultUnitOfWork uow = new DefaultUnitOfWork(testExchange);
    final FcrepoTransactionManager txMgr = new FcrepoTransactionManager();
    txMgr.setBaseUrl(baseUrl);

    testEndpoint.setTransactionManager(txMgr);

    when(mockClient2.post(eq(beginUri))).thenReturn(mockPostBuilder2);
    when(mockClient2.post(eq(commitUri))).thenReturn(mockPostBuilder2);

    init();
    TestUtils.setField(txMgr, "fcrepoClient", mockClient2);

    uow.beginTransactedBy((Object)tx);

    testExchange.getIn().setHeader(FCREPO_IDENTIFIER, path);
    testExchange.setUnitOfWork(uow);

    when(mockPostBuilder2.perform()).thenReturn(
            new FcrepoResponse(beginUri, 201, singletonMap("Location", singletonList(baseUrl + "/" + tx)), null));
    when(mockPostBuilder3.perform()).thenReturn(
            new FcrepoResponse(commitUri, 201, emptyMap(), null));

    when(mockHeadBuilder.perform()).thenReturn(new FcrepoResponse(uri, 200, emptyMap(), null));
    when(mockClient.get(eq(uri))).thenReturn(mockGetBuilder2);
    when(mockClient.get(eq(uri2))).thenReturn(mockGetBuilder3);

    when(mockGetBuilder2.perform()).thenReturn(
        new FcrepoResponse(uri, status, singletonMap(CONTENT_TYPE, singletonList(TestUtils.RDF_XML)), body));
    when(mockGetBuilder3.perform()).thenReturn(
        new FcrepoResponse(uri2, status, singletonMap(CONTENT_TYPE, singletonList(N_TRIPLES)), body2));

    testProducer.process(testExchange);

    assertEquals(status, testExchange.getIn().getHeader(HTTP_RESPONSE_CODE));
    assertEquals(TestUtils.RDF_XML, testExchange.getIn().getHeader(CONTENT_TYPE, String.class));
    assertEquals(TestUtils.rdfXml, testExchange.getIn().getBody(String.class));

    testExchange.getIn().setHeader(HTTP_METHOD, "GET");
    testExchange.getIn().setHeader(ACCEPT_CONTENT_TYPE, N_TRIPLES);
    testExchange.getIn().setHeader(FCREPO_IDENTIFIER, path2);
    testExchange.setUnitOfWork(uow);
    testProducer.process(testExchange);
    assertEquals(status, testExchange.getIn().getHeader(HTTP_RESPONSE_CODE));
    assertEquals(N_TRIPLES, testExchange.getIn().getHeader(CONTENT_TYPE, String.class));
    assertEquals(TestUtils.rdfTriples, testExchange.getIn().getBody(String.class));
}

@Test (expected = RuntimeException.class)
public void testTransactedProducerWithError() throws Exception {
    final String baseUrl = "http://localhost:8080/rest";
    final String path = "/transact";
    final String path2 = "/transact2";
    final String tx = "tx:12345";
    final URI uri = create(baseUrl + "/" + tx + path);
    final URI uri2 = create(baseUrl + "/" + tx + path2);
    final URI commitUri = URI.create(baseUrl + "/" + tx + FcrepoConstants.COMMIT);
    final URI beginUri = URI.create(baseUrl + FcrepoConstants.TRANSACTION);
    final int status = 200;
    final ByteArrayInputStream body = new ByteArrayInputStream(TestUtils.rdfXml.getBytes());
    final ByteArrayInputStream body2 = new ByteArrayInputStream(TestUtils.rdfTriples.getBytes());
    final DefaultUnitOfWork uow = new DefaultUnitOfWork(testExchange);
    final FcrepoTransactionManager txMgr = new FcrepoTransactionManager();
    txMgr.setBaseUrl(baseUrl);

    testEndpoint.setTransactionManager(txMgr);

    init();
    TestUtils.setField(txMgr, "fcrepoClient", mockClient2);

    uow.beginTransactedBy((Object)tx);

    testExchange.getIn().setHeader(FCREPO_IDENTIFIER, path);
    testExchange.setUnitOfWork(uow);

    when(mockClient2.post(eq(beginUri))).thenReturn(mockPostBuilder2);
    when(mockClient2.post(eq(commitUri))).thenReturn(mockPostBuilder3);
    when(mockPostBuilder2.perform()).thenReturn(
            new FcrepoResponse(beginUri, 201, singletonMap("Location", singletonList(baseUrl + "/" + tx)), null));
    when(mockPostBuilder3.perform()).thenReturn(
            new FcrepoResponse(commitUri, 201, emptyMap(), null));

    when(mockHeadBuilder.perform()).thenReturn(new FcrepoResponse(uri, 200, emptyMap(), null));
    when(mockClient.get(eq(uri))).thenReturn(mockGetBuilder2);
    when(mockClient.get(eq(uri2))).thenReturn(mockGetBuilder3);
    when(mockGetBuilder2.perform()).thenReturn(
        new FcrepoResponse(uri, status, singletonMap(CONTENT_TYPE, singletonList(TestUtils.RDF_XML)), body));
    when(mockGetBuilder3.perform()).thenThrow(
        new FcrepoOperationFailedException(uri2, 400, "Bad Request"));

    testProducer.process(testExchange);

    assertEquals(status, testExchange.getIn().getHeader(HTTP_RESPONSE_CODE));
    assertEquals(TestUtils.RDF_XML, testExchange.getIn().getHeader(CONTENT_TYPE, String.class));
    assertEquals(TestUtils.rdfXml, testExchange.getIn().getBody(String.class));

    testExchange.getIn().setHeader(HTTP_METHOD, "GET");
    testExchange.getIn().setHeader(ACCEPT_CONTENT_TYPE, N_TRIPLES);
    testExchange.getIn().setHeader(FCREPO_IDENTIFIER, path2);
    testExchange.setUnitOfWork(uow);
    testProducer.process(testExchange);
}