diff --git a/build.gradle b/build.gradle index 02d7546f73..140e8bc471 100644 --- a/build.gradle +++ b/build.gradle @@ -79,7 +79,7 @@ subprojects { subproject -> } ext { - activeMqVersion = '5.13.2' + activeMqVersion = '5.13.4' aspectjVersion = '1.8.9' apacheSshdVersion = '0.14.0' boonVersion = '0.33' @@ -102,7 +102,7 @@ subprojects { subproject -> javaxActivationVersion = '1.1.1' javaxMailVersion = '1.5.5' jedisVersion = '2.7.3' - jmsApiVersion = '1.1-rev-1' + jmsApiVersion = '2.0.1' jpa21ApiVersion = '1.0.0.Final' jpaApiVersion = '2.1.1' jrubyVersion = '1.7.23' @@ -114,11 +114,10 @@ subprojects { subproject -> log4jVersion = '1.2.17' mockitoVersion = '1.10.19' mysqlVersion = '5.1.34' - nettyVersion = '4.0.27.Final' - openJpaVersion = '2.4.1' + nettyVersion = '4.1.4.Final' pahoMqttClientVersion = '1.0.2' postgresVersion = '9.1-901-1.jdbc4' - reactorVersion = '2.5.0.BUILD-SNAPSHOT' + reactorVersion = '3.0.0.BUILD-SNAPSHOT' romeToolsVersion = '1.6.0' servletApiVersion = '3.1.0' slf4jVersion = "1.7.21" @@ -132,7 +131,7 @@ subprojects { subproject -> springSecurityVersion = '4.1.0.RELEASE' springSocialTwitterVersion = '1.1.2.RELEASE' springRetryVersion = '1.1.2.RELEASE' - springVersion = project.hasProperty('springVersion') ? project.springVersion : '4.3.2.RELEASE' + springVersion = project.hasProperty('springVersion') ? project.springVersion : '5.0.0.BUILD-SNAPSHOT' springWsVersion = '2.3.0.RELEASE' xmlUnitVersion = '1.6' xstreamVersion = '1.4.7' @@ -421,7 +420,7 @@ project('spring-integration-jms') { dependencies { compile project(":spring-integration-core") compile "org.springframework:spring-jms:$springVersion" - compile ("javax.jms:jms-api:$jmsApiVersion", provided) + compile ("javax.jms:javax.jms-api:$jmsApiVersion", provided) testCompile("org.apache.activemq:activemq-broker:$activeMqVersion") testCompile "org.springframework:spring-oxm:$springVersion" } @@ -452,10 +451,6 @@ project('spring-integration-jpa') { testCompile "org.hibernate:hibernate-entitymanager:$hibernateVersion" - testCompile ("org.apache.openjpa:openjpa:$openJpaVersion") { - exclude group: 'org.apache.geronimo.specs' - } - testRuntime "org.hibernate.javax.persistence:hibernate-jpa-2.1-api:$jpa21ApiVersion" testRuntime "org.eclipse.persistence:org.eclipse.persistence.jpa:$eclipseLinkVersion" testRuntime "org.springframework:spring-instrument:$springVersion" @@ -463,7 +458,7 @@ project('spring-integration-jpa') { tasks.withType(Test).matching {it.name ==~ /(springIo.+)|(test)|(testAll)/}.all { jvmArgs classpath.files.findAll { - it.name ==~ /(spring-instrument.+)|(openjpa-\d+\.\d+\.\d+\.jar)/}.collect{"-javaagent:$it" + it.name ==~ /(spring-instrument.+)/}.collect{"-javaagent:$it" } } @@ -567,6 +562,7 @@ project('spring-integration-stomp') { description = 'Spring Integration STOMP Support' dependencies { compile project(":spring-integration-core") + compile ("org.springframework:spring-websocket:$springVersion", optional) testCompile project(":spring-integration-websocket") @@ -576,7 +572,10 @@ project('spring-integration-stomp') { testCompile "org.apache.tomcat.embed:tomcat-embed-websocket:$tomcatVersion" testRuntime "org.apache.tomcat.embed:tomcat-embed-logging-log4j:$tomcatVersion" - testRuntime "io.projectreactor:reactor-net:$reactorVersion" + testRuntime("io.projectreactor:reactor-core:2.0.8.RELEASE") { + force = true // enforce 2.0.x + } + testRuntime("io.projectreactor:reactor-net:2.0.8.RELEASE") testRuntime "io.netty:netty-all:$nettyVersion" } } @@ -654,7 +653,7 @@ project('spring-integration-ws') { testCompile ("org.springframework:spring-jms:$springVersion") { exclude group: 'org.springframework' } - testCompile "javax.jms:jms-api:$jmsApiVersion" + testCompile "javax.jms:javax.jms-api:$jmsApiVersion" testCompile "org.igniterealtime.smack:smack-tcp:$smackVersion" testCompile "org.igniterealtime.smack:smack-java7:$smackVersion" testCompile "org.igniterealtime.smack:smack-extensions:$smackVersion" diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/ReactiveChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/ReactiveChannel.java index 67efd16887..1e5f58eab5 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/ReactiveChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/ReactiveChannel.java @@ -19,15 +19,13 @@ package org.springframework.integration.channel; import org.reactivestreams.Processor; import org.reactivestreams.Publisher; import org.reactivestreams.Subscriber; -import org.reactivestreams.Subscription; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.util.Assert; +import reactor.core.publisher.BlockingSink; import reactor.core.publisher.DirectProcessor; -import reactor.core.subscriber.BaseSubscriber; -import reactor.core.subscriber.SubmissionEmitter; /** * @author Artem Bilan @@ -37,7 +35,7 @@ public class ReactiveChannel implements MessageChannel, Publisher> { private final Processor, Message> processor; - private final SubmissionEmitter> emitter; + private final BlockingSink> sink; public ReactiveChannel() { this(DirectProcessor.create()); @@ -46,24 +44,7 @@ public class ReactiveChannel implements MessageChannel, Publisher> { public ReactiveChannel(Processor, Message> processor) { Assert.notNull(processor, "'processor' must not be null"); this.processor = processor; - this.emitter = SubmissionEmitter.create(processor); - } - - public Subscriber> asSubscriber() { - return new BaseSubscriber>() { - - @Override - public void onSubscribe(Subscription subscription) { - Assert.notNull(subscription, "'subscription' must not be null"); - subscription.request(Long.MAX_VALUE); - } - - @Override - public void onNext(Message message) { - send(message); - } - - }; + this.sink = BlockingSink.create(this.processor); } @Override @@ -73,7 +54,7 @@ public class ReactiveChannel implements MessageChannel, Publisher> { @Override public boolean send(Message message, long timeout) { - return this.emitter.submit(message, timeout) > -1; + return this.sink.submit(message, timeout) > -1; } @Override diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java index 583f02f21c..6377c82d4f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java @@ -40,7 +40,7 @@ import org.springframework.integration.context.IntegrationObjectSupport; import org.springframework.integration.endpoint.AbstractEndpoint; import org.springframework.integration.endpoint.EventDrivenConsumer; import org.springframework.integration.endpoint.PollingConsumer; -import org.springframework.integration.endpoint.ReactiveEndpoint; +import org.springframework.integration.endpoint.ReactiveConsumer; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; import org.springframework.integration.handler.advice.HandleMessageAdvice; import org.springframework.integration.scheduling.PollerMetadata; @@ -55,8 +55,6 @@ import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; import org.springframework.util.StringUtils; -import reactor.core.subscriber.Subscribers; - /** * @author Mark Fisher @@ -290,14 +288,12 @@ public class ConsumerEndpointFactoryBean this.endpoint = pollingConsumer; } else { - Subscriber> subscriber; if (this.handler instanceof Subscriber) { - subscriber = (Subscriber>) this.handler; + this.endpoint = new ReactiveConsumer(channel, (Subscriber>) this.handler); } else { - subscriber = Subscribers.consumer(this.handler::handleMessage); + this.endpoint = new ReactiveConsumer(channel, this.handler::handleMessage); } - this.endpoint = new ReactiveEndpoint(channel, subscriber); } this.endpoint.setBeanName(this.beanName); this.endpoint.setBeanFactory(this.beanFactory); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java index 423791eeef..f8c40a911b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java @@ -55,7 +55,7 @@ import org.springframework.integration.endpoint.AbstractEndpoint; import org.springframework.integration.endpoint.AbstractPollingEndpoint; import org.springframework.integration.endpoint.EventDrivenConsumer; import org.springframework.integration.endpoint.PollingConsumer; -import org.springframework.integration.endpoint.ReactiveEndpoint; +import org.springframework.integration.endpoint.ReactiveConsumer; import org.springframework.integration.endpoint.SourcePollingChannelAdapter; import org.springframework.integration.handler.AbstractMessageProducingHandler; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; @@ -80,7 +80,7 @@ import org.springframework.util.CollectionUtils; import org.springframework.util.ObjectUtils; import org.springframework.util.StringUtils; -import reactor.core.subscriber.Subscribers; +import reactor.core.publisher.DirectProcessor; /** * Base class for Method-level annotation post-processors. @@ -326,9 +326,11 @@ public abstract class AbstractMethodAnnotationPostProcessor> directProcessor = DirectProcessor.create(); + directProcessor.doOnNext(handler::handleMessage); + subscriber = directProcessor; } - endpoint = new ReactiveEndpoint(inputChannel, subscriber); + endpoint = new ReactiveConsumer(inputChannel, subscriber); } else { endpoint = new EventDrivenConsumer((SubscribableChannel) inputChannel, handler); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveEndpoint.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveConsumer.java similarity index 76% rename from spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveEndpoint.java rename to spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveConsumer.java index b57ba75620..3d21e9736d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveEndpoint.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveConsumer.java @@ -16,6 +16,8 @@ package org.springframework.integration.endpoint; +import java.util.function.Consumer; + import org.reactivestreams.Publisher; import org.reactivestreams.Subscriber; import org.reactivestreams.Subscription; @@ -31,47 +33,58 @@ import org.springframework.util.Assert; import org.springframework.util.ErrorHandler; import reactor.core.publisher.DirectProcessor; -import reactor.core.subscriber.SubscriberBarrier; -import reactor.core.util.Exceptions; +import reactor.core.publisher.Flux; /** * @author Artem Bilan * @since 5.0 */ -public class ReactiveEndpoint extends AbstractEndpoint { +public class ReactiveConsumer extends AbstractEndpoint { - private final Publisher> publisher; + private final Subscriber> subscriber; - private final SubscriberBarrier, Message> subscriber; + private final Consumer> consumer; + + private volatile Flux> publisher; + + private volatile Subscription subscription; private ErrorHandler errorHandler; - @SuppressWarnings("unchecked") - public ReactiveEndpoint(MessageChannel inputChannel, Subscriber> subscriber) { - Assert.notNull(inputChannel); + public ReactiveConsumer(MessageChannel inputChannel, Subscriber> subscriber) { + this(inputChannel, subscriber, null); Assert.notNull(subscriber); + + } + + public ReactiveConsumer(MessageChannel inputChannel, Consumer> consumer) { + this(inputChannel, null, consumer); + Assert.notNull(consumer); + } + + @SuppressWarnings("unchecked") + private ReactiveConsumer(MessageChannel inputChannel, Subscriber> subscriber, + Consumer> consumer) { + Assert.notNull(inputChannel); + + Publisher> publisher; if (inputChannel instanceof Publisher) { - this.publisher = (Publisher>) inputChannel; + publisher = (Publisher>) inputChannel; } else { - this.publisher = adaptToPublisher(inputChannel); + publisher = adaptToPublisher(inputChannel); } - this.subscriber = new SubscriberBarrier, Message>(subscriber) { - @Override - protected void doNext(Message message) { - try { - super.doNext(message); - } - catch (Exception e) { - Exceptions.throwIfFatal(e); - ReactiveEndpoint.this.errorHandler.handleError(e); - } - } + this.publisher = Flux.from(publisher) + .log() + .retry() + .doOnError(t -> this.errorHandler.handleError(t)) // NPE if method reference + .doOnSubscribe(s -> this.subscription = s); - }; + this.subscriber = subscriber; + this.consumer = consumer; } public void setErrorHandler(ErrorHandler errorHandler) { @@ -90,12 +103,19 @@ public class ReactiveEndpoint extends AbstractEndpoint { @Override protected void doStart() { - this.publisher.subscribe(this.subscriber); + if (this.subscriber != null) { + this.publisher.subscribe(this.subscriber); + } + else { + this.publisher.subscribe(this.consumer); + } } @Override protected void doStop() { - this.subscriber.cancel(); + if (this.subscription != null) { + this.subscription.cancel(); + } } private Publisher> adaptToPublisher(MessageChannel inputChannel) { @@ -125,7 +145,7 @@ public class ReactiveEndpoint extends AbstractEndpoint { private final DirectProcessor> delegate = DirectProcessor.create(); - private final MessageHandler subscriberAdapter = this.delegate.connectEmitter()::accept; + private final MessageHandler subscriberAdapter = this.delegate.connectSink()::accept; private final SubscribableChannel channel; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java index 5fdc5c3cae..0b84229f77 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java @@ -37,7 +37,7 @@ import org.springframework.messaging.MessageHandlingException; import org.springframework.messaging.MessagingException; import org.springframework.util.Assert; -import reactor.core.util.Exceptions; +import reactor.core.Exceptions; /** * Base class for MessageHandler implementations that provides basic validation @@ -168,7 +168,7 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport im @Override public void onComplete() { - + System.out.println("onComplete()"); } protected abstract void handleMessageInternal(Message message) throws Exception; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/transformer/MapToObjectTransformer.java b/spring-integration-core/src/main/java/org/springframework/integration/transformer/MapToObjectTransformer.java index 54a7be5c87..5ed1aa8871 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/transformer/MapToObjectTransformer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/transformer/MapToObjectTransformer.java @@ -78,7 +78,7 @@ public class MapToObjectTransformer extends AbstractPayloadTransformer @Override protected Object transformPayload(Map payload) throws Exception { Object target = (this.targetClass != null) - ? BeanUtils.instantiate(this.targetClass) + ? BeanUtils.instantiateClass(this.targetClass) : this.getBeanFactory().getBean(this.targetBeanName); DataBinder binder = new DataBinder(target); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveEndpointTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveConsumerTests.java similarity index 59% rename from spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveEndpointTests.java rename to spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveConsumerTests.java index 58f69c0bda..1c3bf28e69 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveEndpointTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveConsumerTests.java @@ -16,6 +16,7 @@ package org.springframework.integration.channel.reactive; +import static org.hamcrest.Matchers.equalTo; import static org.hamcrest.Matchers.instanceOf; import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; @@ -32,9 +33,11 @@ import org.junit.Test; import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.BeanFactory; +import org.springframework.beans.factory.config.ConfigurableBeanFactory; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.ReactiveChannel; -import org.springframework.integration.endpoint.ReactiveEndpoint; +import org.springframework.integration.config.ConsumerEndpointFactoryBean; +import org.springframework.integration.endpoint.ReactiveConsumer; import org.springframework.integration.handler.MethodInvokingMessageHandler; import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.Message; @@ -43,18 +46,17 @@ import org.springframework.messaging.MessageHandler; import org.springframework.messaging.support.GenericMessage; import reactor.core.publisher.EmitterProcessor; -import reactor.core.test.TestSubscriber; +import reactor.test.TestSubscriber; /** * @author Artem Bilan * @since 5.0 */ -public class ReactiveEndpointTests { +public class ReactiveConsumerTests { @Test - public void testReactiveEndpointReactiveChannel() throws InterruptedException { - ReactiveChannel testChannel = - new ReactiveChannel(EmitterProcessor.create(false)); + public void testReactiveConsumerReactiveChannel() throws InterruptedException { + ReactiveChannel testChannel = new ReactiveChannel(EmitterProcessor.create(false)); List> result = new LinkedList<>(); CountDownLatch stopLatch = new CountDownLatch(2); @@ -66,19 +68,19 @@ public class ReactiveEndpointTests { MethodInvokingMessageHandler testSubscriber = new MethodInvokingMessageHandler(messageHandler, (String) null); - ReactiveEndpoint reactiveEndpoint = new ReactiveEndpoint(testChannel, testSubscriber); - reactiveEndpoint.setBeanFactory(mock(BeanFactory.class)); - reactiveEndpoint.afterPropertiesSet(); - reactiveEndpoint.start(); + ReactiveConsumer reactiveConsumer = new ReactiveConsumer(testChannel, testSubscriber); + reactiveConsumer.setBeanFactory(mock(BeanFactory.class)); + reactiveConsumer.afterPropertiesSet(); + reactiveConsumer.start(); Message testMessage = new GenericMessage<>("test"); testChannel.send(testMessage); - reactiveEndpoint.stop(); + reactiveConsumer.stop(); testChannel.send(testMessage); - reactiveEndpoint.start(); + reactiveConsumer.start(); Message testMessage2 = new GenericMessage<>("test2"); testChannel.send(testMessage2); @@ -89,15 +91,15 @@ public class ReactiveEndpointTests { @Test - public void testReactiveEndpointDirectChannel() { + public void testReactiveConsumerDirectChannel() { DirectChannel testChannel = new DirectChannel(); - TestSubscriber> testSubscriber = new TestSubscriber<>(); + TestSubscriber> testSubscriber = TestSubscriber.create(); - ReactiveEndpoint reactiveEndpoint = new ReactiveEndpoint(testChannel, testSubscriber); - reactiveEndpoint.setBeanFactory(mock(BeanFactory.class)); - reactiveEndpoint.afterPropertiesSet(); - reactiveEndpoint.start(); + ReactiveConsumer reactiveConsumer = new ReactiveConsumer(testChannel, testSubscriber); + reactiveConsumer.setBeanFactory(mock(BeanFactory.class)); + reactiveConsumer.afterPropertiesSet(); + reactiveConsumer.start(); Message testMessage = new GenericMessage<>("test"); testChannel.send(testMessage); @@ -108,7 +110,7 @@ public class ReactiveEndpointTests { testSubscriber.assertValues(testMessage); - reactiveEndpoint.stop(); + reactiveConsumer.stop(); try { testChannel.send(testMessage); @@ -121,7 +123,7 @@ public class ReactiveEndpointTests { new DirectFieldAccessor(testSubscriber).setPropertyValue("s", null); TestUtils.getPropertyValue(testSubscriber, "values", List.class).clear(); - reactiveEndpoint.start(); + reactiveConsumer.start(); testSubscriber.request(1); @@ -130,12 +132,45 @@ public class ReactiveEndpointTests { testChannel.send(testMessage); testSubscriber.assertValues(testMessage); + } + + @Test + public void testReactiveConsumerViaConsumerEndpointFactoryBean() throws Exception { + ReactiveChannel testChannel = new ReactiveChannel(); + + List> result = new LinkedList<>(); + CountDownLatch stopLatch = new CountDownLatch(3); + + MessageHandler messageHandler = m -> { + result.add(m); + stopLatch.countDown(); + }; + + ConsumerEndpointFactoryBean endpointFactoryBean = new ConsumerEndpointFactoryBean(); + endpointFactoryBean.setBeanFactory(mock(ConfigurableBeanFactory.class)); + endpointFactoryBean.setInputChannel(testChannel); + endpointFactoryBean.setHandler(messageHandler); + endpointFactoryBean.setBeanName("reactiveConsumer"); + endpointFactoryBean.afterPropertiesSet(); + endpointFactoryBean.start(); + + Message testMessage = new GenericMessage<>("test"); + testChannel.send(testMessage); + + endpointFactoryBean.stop(); testChannel.send(testMessage); - testSubscriber.assertError(IllegalStateException.class); - testSubscriber.assertErrorMessage("Can't deliver value due to lack of requests"); + endpointFactoryBean.start(); + Message testMessage2 = new GenericMessage<>("test2"); + + testChannel.send(testMessage2); + testChannel.send(testMessage2); + + assertTrue(stopLatch.await(10, TimeUnit.SECONDS)); + assertThat(result.size(), equalTo(3)); + assertThat(result, Matchers.>contains(testMessage, testMessage2, testMessage2)); } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/GatewayParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/GatewayParserTests.java index b84799eba9..ce29fa8d4b 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/GatewayParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/GatewayParserTests.java @@ -175,7 +175,7 @@ public class GatewayParserTests { this.startResponder(requestChannel, replyChannel); TestService service = context.getBean("promise", TestService.class); Mono> result = service.promise("foo"); - Message reply = result.get(Duration.ofSeconds(1)); + Message reply = result.block(Duration.ofSeconds(1)); assertEquals("foo", reply.getPayload()); assertNotNull(TestUtils.getPropertyValue(context.getBean("&promise"), "asyncExecutor")); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.java b/spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.java index 152f6af357..7208aeac08 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.java @@ -128,6 +128,7 @@ import org.springframework.test.context.support.AnnotationConfigContextLoader; import org.springframework.util.MultiValueMap; import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; /** * @author Artem Bilan @@ -620,7 +621,7 @@ public class EnableIntegrationTests { Flux.just("1", "2", "3", "4", "5") .map(Integer::parseInt) .flatMap(this.testGateway::multiply) - .toList() + .collectList() .subscribe(integers -> { ref.set(integers); consumeLatch.countDown(); @@ -1334,7 +1335,7 @@ public class EnableIntegrationTests { void sendAsync(String payload); @Gateway(requestChannel = "promiseChannel") - org.reactivestreams.Publisher multiply(Integer value); + Mono multiply(Integer value); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/core/AsyncMessagingTemplateTests.java b/spring-integration-core/src/test/java/org/springframework/integration/core/AsyncMessagingTemplateTests.java index 8a69d44537..1f999aac0a 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/core/AsyncMessagingTemplateTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/core/AsyncMessagingTemplateTests.java @@ -391,7 +391,7 @@ public class AsyncMessagingTemplateTests { @Test(expected = TimeoutException.class) public void timeoutException() throws Exception { DirectChannel channel = new DirectChannel(); - channel.subscribe(new EchoHandler(200)); + channel.subscribe(new EchoHandler(10000)); AsyncMessagingTemplate template = new AsyncMessagingTemplate(); template.setDefaultDestination(channel); Future> result = template.asyncSendAndReceive(MessageBuilder.withPayload("test").build()); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java b/spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java index 8e6d96cef4..39b38067d0 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/gateway/AsyncGatewayTests.java @@ -238,7 +238,7 @@ public class AsyncGatewayTests { proxyFactory.afterPropertiesSet(); TestEchoService service = (TestEchoService) proxyFactory.getObject(); Mono> promise = service.returnMessagePromise("foo"); - Object result = promise.get(Duration.ofSeconds(10)); + Object result = promise.block(Duration.ofSeconds(10)); assertEquals("foobar", ((Message) result).getPayload()); } @@ -254,7 +254,7 @@ public class AsyncGatewayTests { proxyFactory.afterPropertiesSet(); TestEchoService service = (TestEchoService) proxyFactory.getObject(); Mono promise = service.returnStringPromise("foo"); - Object result = promise.get(Duration.ofSeconds(10)); + Object result = promise.block(Duration.ofSeconds(10)); assertEquals("foobar", result); } @@ -270,7 +270,7 @@ public class AsyncGatewayTests { proxyFactory.afterPropertiesSet(); TestEchoService service = (TestEchoService) proxyFactory.getObject(); Mono promise = service.returnSomethingPromise("foo"); - Object result = promise.get(Duration.ofSeconds(10)); + Object result = promise.block(Duration.ofSeconds(10)); assertNotNull(result); assertEquals("foobar", result); } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java b/spring-integration-file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java index 06cce46b1e..1578010356 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java @@ -49,7 +49,6 @@ import org.springframework.integration.core.MessageSource; import org.springframework.integration.file.filters.AcceptOnceFileListFilter; import org.springframework.integration.file.filters.FileListFilter; import org.springframework.integration.file.filters.ResettableFileListFilter; -import org.springframework.lang.UsesJava7; import org.springframework.messaging.Message; import org.springframework.messaging.MessagingException; import org.springframework.util.Assert; @@ -406,7 +405,6 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement } } - @UsesJava7 public enum WatchEventType { CREATE(StandardWatchEventKinds.ENTRY_CREATE), @@ -423,7 +421,6 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement } - @UsesJava7 private class WatchServiceDirectoryScanner extends DefaultDirectoryScanner implements Lifecycle { private final ConcurrentMap pathKeys = new ConcurrentHashMap(); diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/FileWritingMessageHandler.java b/spring-integration-file/src/main/java/org/springframework/integration/file/FileWritingMessageHandler.java index f633e75c92..b46d1e5bef 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/FileWritingMessageHandler.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/FileWritingMessageHandler.java @@ -53,7 +53,6 @@ import org.springframework.integration.support.locks.DefaultLockRegistry; import org.springframework.integration.support.locks.LockRegistry; import org.springframework.integration.support.locks.PassThruLockRegistry; import org.springframework.integration.util.WhileLockedProcessor; -import org.springframework.lang.UsesJava7; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHandler; import org.springframework.messaging.MessageHandlingException; @@ -871,7 +870,6 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand return (nioFilesPresent && filesMove(source, target)) || source.renameTo(target); } - @UsesJava7 private static boolean filesMove(File source, File target) throws IOException { Files.move(source.toPath(), target.toPath(), StandardCopyOption.REPLACE_EXISTING); return true; diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/WatchServiceDirectoryScanner.java b/spring-integration-file/src/main/java/org/springframework/integration/file/WatchServiceDirectoryScanner.java index 50376c8c58..d6798b0561 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/WatchServiceDirectoryScanner.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/WatchServiceDirectoryScanner.java @@ -37,7 +37,6 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.context.SmartLifecycle; -import org.springframework.lang.UsesJava7; import org.springframework.util.Assert; /** @@ -65,7 +64,6 @@ import org.springframework.util.Assert; * */ @Deprecated -@UsesJava7 @SuppressWarnings("deprecation") public class WatchServiceDirectoryScanner extends DefaultDirectoryScanner implements SmartLifecycle { diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/StubConnection.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/StubConnection.java index c1c6a52164..c8d762ccb0 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/StubConnection.java +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/StubConnection.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2007 the original author or authors. + * Copyright 2002-2016 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -28,6 +28,7 @@ import javax.jms.Topic; /** * @author Mark Fisher + * @author Artem Bilan */ public class StubConnection implements Connection { @@ -38,46 +39,78 @@ public class StubConnection implements Connection { this.messageText = messageText; } - + @Override public void close() throws JMSException { } + @Override public ConnectionConsumer createConnectionConsumer(Destination destination, String messageSelector, ServerSessionPool sessionPool, int maxMessages) throws JMSException { return null; } + @Override public ConnectionConsumer createDurableConnectionConsumer(Topic topic, String subscriptionName, String messageSelector, ServerSessionPool sessionPool, int maxMessages) throws JMSException { return null; } + @Override public Session createSession(boolean transacted, int acknowledgeMode) throws JMSException { return new StubSession(this.messageText); } + @Override public String getClientID() throws JMSException { return null; } + @Override public ExceptionListener getExceptionListener() throws JMSException { return null; } + @Override public ConnectionMetaData getMetaData() throws JMSException { return null; } + @Override public void setClientID(String clientID) throws JMSException { } + @Override public void setExceptionListener(ExceptionListener listener) throws JMSException { } + @Override public void start() throws JMSException { } + @Override public void stop() throws JMSException { } + @Override + public Session createSession(int sessionMode) throws JMSException { + return new StubSession(this.messageText); + } + + @Override + public Session createSession() throws JMSException { + return new StubSession(this.messageText); + } + + @Override + public ConnectionConsumer createSharedConnectionConsumer(Topic topic, String subscriptionName, + String messageSelector, ServerSessionPool sessionPool, int maxMessages) throws JMSException { + return null; + } + + @Override + public ConnectionConsumer createSharedDurableConnectionConsumer(Topic topic, String subscriptionName, + String messageSelector, ServerSessionPool sessionPool, int maxMessages) throws JMSException { + return null; + } + } diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/StubProducer.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/StubProducer.java index 59aaa5b709..34064c97fb 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/StubProducer.java +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/StubProducer.java @@ -16,6 +16,7 @@ package org.springframework.integration.jms; +import javax.jms.CompletionListener; import javax.jms.Destination; import javax.jms.InvalidDestinationException; import javax.jms.JMSException; @@ -31,6 +32,8 @@ public class StubProducer implements MessageProducer { private Destination staticDestination; + private long deliveryDelay; + public StubProducer(Destination destination) { this.staticDestination = destination; @@ -118,4 +121,37 @@ public class StubProducer implements MessageProducer { public void setTimeToLive(long timeToLive) throws JMSException { } + @Override + public void setDeliveryDelay(long deliveryDelay) throws JMSException { + this.deliveryDelay = deliveryDelay; + } + + @Override + public long getDeliveryDelay() throws JMSException { + return this.deliveryDelay; + } + + @Override + public void send(Message message, CompletionListener completionListener) throws JMSException { + send(message); + } + + @Override + public void send(Message message, int deliveryMode, int priority, long timeToLive, + CompletionListener completionListener) throws JMSException { + send(message); + } + + @Override + public void send(Destination destination, Message message, CompletionListener completionListener) + throws JMSException { + send(destination, message); + } + + @Override + public void send(Destination destination, Message message, int deliveryMode, int priority, long timeToLive, + CompletionListener completionListener) throws JMSException { + send(destination, message); + } + } diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/StubSession.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/StubSession.java index 11211aab17..292497c683 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/StubSession.java +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/StubSession.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2009 the original author or authors. + * Copyright 2002-2016 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -39,6 +39,7 @@ import javax.jms.TopicSubscriber; /** * @author Mark Fisher + * @author Artem Bilan */ public class StubSession implements Session { @@ -49,120 +50,181 @@ public class StubSession implements Session { this.messageText = messageText; } - + @Override public void close() throws JMSException { } + @Override public void commit() throws JMSException { } + @Override public QueueBrowser createBrowser(Queue queue) throws JMSException { return null; } + @Override public QueueBrowser createBrowser(Queue queue, String messageSelector) throws JMSException { return null; } + @Override public BytesMessage createBytesMessage() throws JMSException { return null; } + @Override public MessageConsumer createConsumer(Destination destination) throws JMSException { return new StubConsumer(this.messageText, null); } + @Override public MessageConsumer createConsumer(Destination destination, String messageSelector) throws JMSException { return new StubConsumer(this.messageText, messageSelector); } + @Override public MessageConsumer createConsumer(Destination destination, String messageSelector, boolean NoLocal) throws JMSException { return new StubConsumer(this.messageText, messageSelector); } + @Override public TopicSubscriber createDurableSubscriber(Topic topic, String name) throws JMSException { return null; } + @Override public TopicSubscriber createDurableSubscriber(Topic topic, String name, String messageSelector, boolean noLocal) throws JMSException { return null; } + @Override public MapMessage createMapMessage() throws JMSException { return null; } + @Override public Message createMessage() throws JMSException { return null; } + @Override public ObjectMessage createObjectMessage() throws JMSException { return null; } + @Override public ObjectMessage createObjectMessage(Serializable object) throws JMSException { return null; } + @Override public MessageProducer createProducer(Destination destination) throws JMSException { return new StubProducer(destination); } + @Override public Queue createQueue(String queueName) throws JMSException { return null; } + @Override public StreamMessage createStreamMessage() throws JMSException { return null; } + @Override public TemporaryQueue createTemporaryQueue() throws JMSException { return null; } + @Override public TemporaryTopic createTemporaryTopic() throws JMSException { return null; } + @Override public TextMessage createTextMessage() throws JMSException { return null; } + @Override public TextMessage createTextMessage(String text) throws JMSException { return new StubTextMessage(text); } + @Override public Topic createTopic(String topicName) throws JMSException { return null; } + @Override public int getAcknowledgeMode() throws JMSException { return 0; } + @Override public MessageListener getMessageListener() throws JMSException { return null; } + @Override public boolean getTransacted() throws JMSException { return false; } + @Override public void recover() throws JMSException { } + @Override public void rollback() throws JMSException { } + @Override public void run() { } + @Override public void setMessageListener(MessageListener listener) throws JMSException { } + @Override public void unsubscribe(String name) throws JMSException { } + @Override + public MessageConsumer createSharedConsumer(Topic topic, String sharedSubscriptionName) throws JMSException { + return new StubConsumer(this.messageText, null); + } + + @Override + public MessageConsumer createSharedConsumer(Topic topic, String sharedSubscriptionName, String messageSelector) throws JMSException { + return new StubConsumer(this.messageText, messageSelector); + } + + @Override + public MessageConsumer createDurableConsumer(Topic topic, String name) throws JMSException { + return new StubConsumer(this.messageText, null); + } + + @Override + public MessageConsumer createDurableConsumer(Topic topic, String name, String messageSelector, boolean noLocal) + throws JMSException { + return new StubConsumer(this.messageText, messageSelector); + } + + @Override + public MessageConsumer createSharedDurableConsumer(Topic topic, String name) throws JMSException { + return new StubConsumer(this.messageText, null); + } + + @Override + public MessageConsumer createSharedDurableConsumer(Topic topic, String name, String messageSelector) + throws JMSException { + return new StubConsumer(this.messageText, messageSelector); + } + } diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/StubTextMessage.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/StubTextMessage.java index e4a5450d1b..58d7d33df7 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/StubTextMessage.java +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/StubTextMessage.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2016 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -27,6 +27,7 @@ import javax.jms.TextMessage; * Stub JMS Message implementation intended for testing purposes only. * * @author Mark Fisher + * @author Artem Bilan */ public class StubTextMessage implements TextMessage { @@ -54,6 +55,8 @@ public class StubTextMessage implements TextMessage { private ConcurrentHashMap properties = new ConcurrentHashMap(); + private long deliveryTime; + public StubTextMessage() { } @@ -62,201 +65,269 @@ public class StubTextMessage implements TextMessage { this.text = text; } - + @Override public String getText() throws JMSException { return this.text; } + @Override public void setText(String text) throws JMSException { this.text = text; } + @Override public void acknowledge() throws JMSException { throw new UnsupportedOperationException(); } + @Override public void clearBody() throws JMSException { this.text = null; } + @Override public void clearProperties() throws JMSException { this.properties.clear(); } + @Override public boolean getBooleanProperty(String name) throws JMSException { Object value = this.properties.get(name); - return (value instanceof Boolean) ? ((Boolean) value).booleanValue() : false; + return (value instanceof Boolean) && (Boolean) value; } + @Override public byte getByteProperty(String name) throws JMSException { Object value = this.properties.get(name); - return (value instanceof Byte) ? ((Byte) value).byteValue() : 0; + return (value instanceof Byte) ? (Byte) value : 0; } + @Override public double getDoubleProperty(String name) throws JMSException { Object value = this.properties.get(name); - return (value instanceof Double) ? ((Double) value).doubleValue() : 0; + return (value instanceof Double) ? (Double) value : 0; } + @Override public float getFloatProperty(String name) throws JMSException { Object value = this.properties.get(name); - return (value instanceof Float) ? ((Float) value).floatValue() : 0; + return (value instanceof Float) ? (Float) value : 0; } + @Override public int getIntProperty(String name) throws JMSException { Object value = this.properties.get(name); - return (value instanceof Integer) ? ((Integer) value).intValue() : 0; + return (value instanceof Integer) ? (Integer) value : 0; } + @Override public String getJMSCorrelationID() throws JMSException { return this.correlationId; } + @Override public byte[] getJMSCorrelationIDAsBytes() throws JMSException { return this.correlationId.getBytes(); } + @Override public int getJMSDeliveryMode() throws JMSException { return this.deliveryMode; } + @Override public Destination getJMSDestination() throws JMSException { return this.destination; } + @Override public long getJMSExpiration() throws JMSException { return this.expiration; } + @Override public String getJMSMessageID() throws JMSException { return this.messageId; } + @Override public int getJMSPriority() throws JMSException { return this.priority; } + @Override public boolean getJMSRedelivered() throws JMSException { return this.redelivered; } + @Override public Destination getJMSReplyTo() throws JMSException { return this.replyTo; } + @Override public long getJMSTimestamp() throws JMSException { return this.timestamp; } + @Override public String getJMSType() throws JMSException { return this.type; } + @Override public long getLongProperty(String name) throws JMSException { Object value = this.properties.get(name); - return (value instanceof Long) ? ((Long) value).longValue() : 0; + return (value instanceof Long) ? (Long) value : 0; } + @Override public Object getObjectProperty(String name) throws JMSException { return this.properties.get(name); } + @Override public Enumeration getPropertyNames() throws JMSException { return this.properties.keys(); } + @Override public short getShortProperty(String name) throws JMSException { Object value = this.properties.get(name); - return (value instanceof Short) ? ((Short) value).shortValue() : 0; + return (value instanceof Short) ? (Short) value : 0; } + @Override public String getStringProperty(String name) throws JMSException { Object value = this.properties.get(name); return (value instanceof String) ? (String) value : null; } + @Override public boolean propertyExists(String name) throws JMSException { return this.properties.containsKey(name); } + @Override public void setBooleanProperty(String name, boolean value) throws JMSException { this.properties.put(name, value); } + @Override public void setByteProperty(String name, byte value) throws JMSException { this.properties.put(name, value); } + @Override public void setDoubleProperty(String name, double value) throws JMSException { this.properties.put(name, value); } + @Override public void setFloatProperty(String name, float value) throws JMSException { this.properties.put(name, value); } + @Override public void setIntProperty(String name, int value) throws JMSException { this.properties.put(name, value); } + @Override public void setJMSCorrelationID(String correlationId) throws JMSException { this.correlationId = correlationId; } + @Override public void setJMSCorrelationIDAsBytes(byte[] correlationID) throws JMSException { this.correlationId = new String(correlationID); } + @Override public void setJMSDeliveryMode(int deliveryMode) throws JMSException { this.deliveryMode = deliveryMode; } + @Override public void setJMSDestination(Destination destination) throws JMSException { this.destination = destination; } + @Override public void setJMSExpiration(long expiration) throws JMSException { this.expiration = expiration; } + @Override public void setJMSMessageID(String id) throws JMSException { this.messageId = id; } + @Override public void setJMSPriority(int priority) throws JMSException { this.priority = priority; } + @Override public void setJMSRedelivered(boolean redelivered) throws JMSException { this.redelivered = redelivered; } + @Override public void setJMSReplyTo(Destination replyTo) throws JMSException { this.replyTo = replyTo; } + @Override public void setJMSTimestamp(long timestamp) throws JMSException { this.timestamp = timestamp; } + @Override public void setJMSType(String type) throws JMSException { this.type = type; } + @Override public void setLongProperty(String name, long value) throws JMSException { this.properties.put(name, value); } + @Override public void setObjectProperty(String name, Object value) throws JMSException { this.properties.put(name, value); } + @Override public void setShortProperty(String name, short value) throws JMSException { this.properties.put(name, value); } + @Override public void setStringProperty(String name, String value) throws JMSException { this.properties.put(name, value); } + @Override + public long getJMSDeliveryTime() throws JMSException { + return this.deliveryTime; + } + + @Override + public void setJMSDeliveryTime(long deliveryTime) throws JMSException { + this.deliveryTime = deliveryTime; + } + + @Override + @SuppressWarnings("unchecked") + public T getBody(Class c) throws JMSException { + return (T) this.text; + } + + @Override + @SuppressWarnings({"unchecked", "rawtypes"}) + public boolean isBodyAssignableTo(Class c) throws JMSException { + return c.isAssignableFrom(String.class); + } + } diff --git a/spring-integration-jpa/src/test/java/org/springframework/integration/jpa/core/OpenJpaJpaOperationsTests-context.xml b/spring-integration-jpa/src/test/java/org/springframework/integration/jpa/core/OpenJpaJpaOperationsTests-context.xml deleted file mode 100644 index 90dc17d2bb..0000000000 --- a/spring-integration-jpa/src/test/java/org/springframework/integration/jpa/core/OpenJpaJpaOperationsTests-context.xml +++ /dev/null @@ -1,16 +0,0 @@ - - - - - - - - - - - - - diff --git a/spring-integration-jpa/src/test/java/org/springframework/integration/jpa/core/OpenJpaJpaOperationsTests.java b/spring-integration-jpa/src/test/java/org/springframework/integration/jpa/core/OpenJpaJpaOperationsTests.java deleted file mode 100644 index d912f07e0f..0000000000 --- a/spring-integration-jpa/src/test/java/org/springframework/integration/jpa/core/OpenJpaJpaOperationsTests.java +++ /dev/null @@ -1,229 +0,0 @@ -/* - * Copyright 2002-2016 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.integration.jpa.core; - -import static org.junit.Assert.assertTrue; - -import java.io.IOException; -import java.sql.SQLException; -import java.text.ParseException; - -import org.apache.openjpa.jdbc.conf.JDBCConfiguration; -import org.apache.openjpa.jdbc.conf.JDBCConfigurationImpl; -import org.apache.openjpa.jdbc.meta.MappingTool; -import org.apache.openjpa.lib.conf.Configurations; -import org.apache.openjpa.lib.util.Options; -import org.apache.openjpa.persistence.InvalidStateException; -import org.apache.openjpa.persistence.PersistenceException; -import org.junit.Assert; -import org.junit.Test; -import org.junit.runner.RunWith; - -import org.springframework.test.context.ContextConfiguration; -import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; - -/** - * Tests the functionality of {@link JpaOperations} and {@link DefaultJpaOperations} - * using the OpenJPA persistence provider. - * - * If you want to run these tests from your IDE, please ensure that you execute - * the tests using a javaagent: - * - * 
- * {@code
- * -javaagent://openjpa-2.1.1.jar
- * }
- *
- * - * @author Gunnar Hillert - * @author Gary Russell - * @since 2.2 - * - */ -@RunWith(SpringJUnit4ClassRunner.class) -@ContextConfiguration -public class OpenJpaJpaOperationsTests extends AbstractJpaOperationsTests { - - @Test - @Override - public void testExecuteUpdateWithNativeQuery() { - - try { - super.testExecuteUpdateWithNativeQuery(); - } - catch (PersistenceException e) { - return; - } - - Assert.fail("Was expecting an Exception as OpenJPA does not support Native SQL Queries with Named Parameters."); - } - - @Test - @Override - public void testExecuteUpdateWithNativeNamedQuery() { - - try { - super.testExecuteUpdateWithNativeNamedQuery(); - } - catch (InvalidStateException e) { - return; - } - - Assert.fail("Was expecting an Exception as OpenJPA does not support Native SQL Queries with Named Parameters."); - } - - /** - * Test method for {@link org.springframework.integration.jpa.core.DefaultJpaOperations#persist(java.lang.Object)}. - * - * http://openjpa.apache.org/builds/1.0.4/apache-openjpa-1.0.4/docs/manual/manual.html#ref_guide_ddl_examples - */ - //@Test - public void testGenerateSchema() { - - String[] arguments = {}; - - Options opts = new Options(); - opts.put("schemaAction", "build"); - opts.put("sql", "build/database/openjpa-h2.sql"); - opts.put("org.springframework.integration.jpa.test.entity.Student", "true"); - - final String[] args = opts.setFromCmdLine(arguments); - - boolean ret = Configurations.runAgainstAllAnchors(opts, - new Configurations.Runnable() { - @Override - public boolean run(Options opts) throws IOException, SQLException { - JDBCConfiguration conf = new JDBCConfigurationImpl(); - conf.setConnectionDriverName("org.h2.Driver"); - conf.setConnectionURL("jdbc:h2:~/test"); - conf.setConnectionUserName("sa"); - conf.setConnectionPassword(""); - - try { - return MappingTool.run(conf, args, opts); - } - finally { - conf.close(); - } - } - }); - assertTrue(ret); - } - - @Test - @Override - public void testExecuteUpdate() { - super.testExecuteUpdate(); - } - - @Test - @Override - public void testExecuteUpdateWithNamedQuery() { - super.testExecuteUpdateWithNamedQuery(); - } - - @Test - @Override - public void testExecuteSelectWithNativeQueryReturningEntityClass() - throws ParseException { - super.testExecuteSelectWithNativeQueryReturningEntityClass(); - } - - @Test - @Override - public void testExecuteSelectWithNativeQuery() throws ParseException { - super.testExecuteSelectWithNativeQuery(); - } - - @Test - @Override - public void testMerge() { - super.testMerge(); - } - - @Test - @Override - public void testMergeCollection() { - super.testMergeCollection(); - } - - @Test - @Override - public void testMergeNullCollection() { - super.testMergeNullCollection(); - } - - @Test - @Override - public void testMergeCollectionWithNullElement() { - super.testMergeCollectionWithNullElement(); - } - - @Test - @Override - public void testPersist() { - super.testPersist(); - } - - @Test - @Override - public void testPersistCollection() { - super.testPersistCollection(); - } - - @Test - @Override - public void testPersistNullCollection() { - super.testPersistNullCollection(); - } - - @Test - @Override - public void testPersistCollectionWithNullElement() { - super.testPersistCollectionWithNullElement(); - } - - @Test - @Override - public void testGetAllStudents() { - super.testGetAllStudents(); - } - - @Test - @Override - public void testGetAllStudentsWithMaxResults() { - super.testGetAllStudentsWithMaxResults(); - } - - @Test - @Override - public void testDeleteInBatch() { - super.testDeleteInBatch(); - } - - @Test - @Override - public void testDelete() { - super.testDelete(); - } - - @Test - @Override - public void testDeleteInBatchWithEmptyCollection() { - super.testDeleteInBatchWithEmptyCollection(); - } -}