SpringCloud/spring-cloud-task
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Added checkstyle

Browse Source
This commit is contained in:
Marcin Grzejszczak
2019-02-03 19:27:07 +01:00
parent 4d0acf120c
commit 60f1e21d03
249 changed files with 7450 additions and 5857 deletions
Download Patch File Download Diff File Expand all files Collapse all files

228
spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/BatchEventAutoConfiguration.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2018 the original author or authors.
* Copyright 2015-2019 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.
@@ -13,6 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.task.batch.listener;
import org.springframework.batch.core.ChunkListener;
@@ -42,16 +43,17 @@ import org.springframework.context.annotation.Lazy;
import org.springframework.messaging.MessageChannel;
/**
* Autoconfigures Spring Batch listeners designed to emit events on the following channels:
* Autoconfigures Spring Batch listeners designed to emit events on the following
* channels.
*
* <ul>
* <li>{@link EventEmittingJobExecutionListener} - job-execution-events</li>
* <li>{@link EventEmittingStepExecutionListener} - step-execution-events</li>
* <li>{@link ChunkListener} - chunk-events</li>
* <li>{@link EventEmittingItemReadListener} - item-read-events</li>
* <li>{@link EventEmittingItemProcessListener} - item-process-events</li>
* <li>{@link EventEmittingItemWriteListener} - item-write-events</li>
* <li>{@link EventEmittingSkipListener} - skip-events</li>
* <li>{@link EventEmittingJobExecutionListener} - job-execution-events</li>
* <li>{@link EventEmittingStepExecutionListener} - step-execution-events</li>
* <li>{@link ChunkListener} - chunk-events</li>
* <li>{@link EventEmittingItemReadListener} - item-read-events</li>
* <li>{@link EventEmittingItemProcessListener} - item-process-events</li>
* <li>{@link EventEmittingItemWriteListener} - item-write-events</li>
* <li>{@link EventEmittingSkipListener} - skip-events</li>
* </ul>
*
* @author Michael Minella
@@ -60,17 +62,46 @@ import org.springframework.messaging.MessageChannel;
*/
@Configuration
@ConditionalOnClass(Job.class)
@ConditionalOnBean(value = { Job.class, TaskLifecycleListener.class })
@ConditionalOnBean({ Job.class, TaskLifecycleListener.class })
// @checkstyle:off
@ConditionalOnProperty(prefix = "spring.cloud.task.batch.events", name = "enabled", havingValue = "true", matchIfMissing = true)
// @checkstyle:on
@AutoConfigureAfter(SimpleTaskAutoConfiguration.class)
public class BatchEventAutoConfiguration {
/**
* Name of the job execution events listener.
*/
public static final String JOB_EXECUTION_EVENTS_LISTENER = "jobExecutionEventsListener";
/**
* Name of the chunk events listener.
*/
public static final String CHUNK_EVENTS_LISTENER = "chunkEventsListener";
/**
* Name of the step execution events listener.
*/
public static final String STEP_EXECUTION_EVENTS_LISTENER = "stepExecutionEventsListener";
/**
* Name of the item read events listener.
*/
public static final String ITEM_READ_EVENTS_LISTENER = "itemReadEventsListener";
/**
* Name of the item write events listener.
*/
public static final String ITEM_WRITE_EVENTS_LISTENER = "itemWriteEventsListener";
/**
* Name of the item process events listener.
*/
public static final String ITEM_PROCESS_EVENTS_LISTENER = "itemProcessEventsListener";
/**
* Name of the skip events listener.
*/
public static final String SKIP_EVENTS_LISTENER = "skipEventsListener";
@Bean
@@ -79,72 +110,44 @@ public class BatchEventAutoConfiguration {
return new TaskBatchEventListenerBeanPostProcessor();
}
@Configuration
@ConditionalOnClass(EnableBinding.class)
@EnableBinding(BatchEventsChannels.class)
@EnableConfigurationProperties(TaskEventProperties.class)
@ConditionalOnMissingBean(name = JOB_EXECUTION_EVENTS_LISTENER)
public static class JobExecutionListenerConfiguration {
@Autowired
private BatchEventsChannels listenerChannels;
@Autowired
private TaskEventProperties taskEventProperties;
@Bean
@Lazy
@ConditionalOnProperty(prefix = "spring.cloud.task.batch.events.job-execution", name = "enabled", havingValue = "true", matchIfMissing = true)
public JobExecutionListener jobExecutionEventsListener() {
return new EventEmittingJobExecutionListener(listenerChannels.jobExecutionEvents(),taskEventProperties.getJobExecutionOrder());
}
@Bean
@ConditionalOnProperty(prefix = "spring.cloud.task.batch.events.step-execution", name = "enabled", havingValue = "true", matchIfMissing = true)
public StepExecutionListener stepExecutionEventsListener() {
return new EventEmittingStepExecutionListener(listenerChannels.stepExecutionEvents(),taskEventProperties.getStepExecutionOrder());
}
@Bean
@Lazy
@ConditionalOnProperty(prefix = "spring.cloud.task.batch.events.chunk", name = "enabled", havingValue = "true", matchIfMissing = true)
public EventEmittingChunkListener chunkEventsListener() {
return new EventEmittingChunkListener(listenerChannels.chunkEvents(),taskEventProperties.getChunkOrder());
}
@Bean
@ConditionalOnProperty(prefix = "spring.cloud.task.batch.events.item-read", name = "enabled", havingValue = "true", matchIfMissing = true)
public ItemReadListener itemReadEventsListener() {
return new EventEmittingItemReadListener(listenerChannels.itemReadEvents(),taskEventProperties.getItemReadOrder());
}
@Bean
@ConditionalOnProperty(prefix = "spring.cloud.task.batch.events.item-write", name = "enabled", havingValue = "true", matchIfMissing = true)
public ItemWriteListener itemWriteEventsListener() {
return new EventEmittingItemWriteListener(listenerChannels.itemWriteEvents(),taskEventProperties.getItemWriteOrder());
}
@Bean
@ConditionalOnProperty(prefix = "spring.cloud.task.batch.events.item-process", name = "enabled", havingValue = "true", matchIfMissing = true)
public ItemProcessListener itemProcessEventsListener() {
return new EventEmittingItemProcessListener(listenerChannels.itemProcessEvents(),taskEventProperties.getItemProcessOrder());
}
@Bean
@ConditionalOnProperty(prefix = "spring.cloud.task.batch.events.skip", name = "enabled", havingValue = "true", matchIfMissing = true)
public SkipListener skipEventsListener() {
return new EventEmittingSkipListener(listenerChannels.skipEvents(),taskEventProperties.getItemProcessOrder());
}
}
/**
* Name of Batch Events channels.
*/
public interface BatchEventsChannels {
/**
* Name of the job execution events channel.
*/
String JOB_EXECUTION_EVENTS = "job-execution-events";
/**
* Name of the step execution events channel.
*/
String STEP_EXECUTION_EVENTS = "step-execution-events";
/**
* Name of the chunk execution events channel.
*/
String CHUNK_EXECUTION_EVENTS = "chunk-events";
/**
* Name of the item read events channel.
*/
String ITEM_READ_EVENTS = "item-read-events";
/**
* Name of the item process events channel.
*/
String ITEM_PROCESS_EVENTS = "item-process-events";
/**
* Name of the item write events channel.
*/
String ITEM_WRITE_EVENTS = "item-write-events";
/**
* Name of the skip events channel.
*/
String SKIP_EVENTS = "skip-events";
@Output(JOB_EXECUTION_EVENTS)
@@ -169,4 +172,93 @@ public class BatchEventAutoConfiguration {
MessageChannel skipEvents();
}
/**
* Configuration for Job Execution Listener.
*/
@Configuration
@ConditionalOnClass(EnableBinding.class)
@EnableBinding(BatchEventsChannels.class)
@EnableConfigurationProperties(TaskEventProperties.class)
@ConditionalOnMissingBean(name = JOB_EXECUTION_EVENTS_LISTENER)
public static class JobExecutionListenerConfiguration {
@Autowired
private BatchEventsChannels listenerChannels;
@Autowired
private TaskEventProperties taskEventProperties;
// @checkstyle:off
@Bean
@Lazy
@ConditionalOnProperty(prefix = "spring.cloud.task.batch.events.job-execution", name = "enabled", havingValue = "true", matchIfMissing = true)
// @checkstyle:on
public JobExecutionListener jobExecutionEventsListener() {
return new EventEmittingJobExecutionListener(
this.listenerChannels.jobExecutionEvents(),
this.taskEventProperties.getJobExecutionOrder());
}
// @checkstyle:off
@Bean
@ConditionalOnProperty(prefix = "spring.cloud.task.batch.events.step-execution", name = "enabled", havingValue = "true", matchIfMissing = true)
// @checkstyle:on
public StepExecutionListener stepExecutionEventsListener() {
return new EventEmittingStepExecutionListener(
this.listenerChannels.stepExecutionEvents(),
this.taskEventProperties.getStepExecutionOrder());
}
// @checkstyle:off
@Bean
@Lazy
@ConditionalOnProperty(prefix = "spring.cloud.task.batch.events.chunk", name = "enabled", havingValue = "true", matchIfMissing = true)
// @checkstyle:on
public EventEmittingChunkListener chunkEventsListener() {
return new EventEmittingChunkListener(this.listenerChannels.chunkEvents(),
this.taskEventProperties.getChunkOrder());
}
// @checkstyle:off
@Bean
@ConditionalOnProperty(prefix = "spring.cloud.task.batch.events.item-read", name = "enabled", havingValue = "true", matchIfMissing = true)
// @checkstyle:on
public ItemReadListener itemReadEventsListener() {
return new EventEmittingItemReadListener(
this.listenerChannels.itemReadEvents(),
this.taskEventProperties.getItemReadOrder());
}
// @checkstyle:off
@Bean
@ConditionalOnProperty(prefix = "spring.cloud.task.batch.events.item-write", name = "enabled", havingValue = "true", matchIfMissing = true)
// @checkstyle:on
public ItemWriteListener itemWriteEventsListener() {
return new EventEmittingItemWriteListener(
this.listenerChannels.itemWriteEvents(),
this.taskEventProperties.getItemWriteOrder());
}
// @checkstyle:off
@Bean
@ConditionalOnProperty(prefix = "spring.cloud.task.batch.events.item-process", name = "enabled", havingValue = "true", matchIfMissing = true)
// @checkstyle:on
public ItemProcessListener itemProcessEventsListener() {
return new EventEmittingItemProcessListener(
this.listenerChannels.itemProcessEvents(),
this.taskEventProperties.getItemProcessOrder());
}
// @checkstyle:off
@Bean
@ConditionalOnProperty(prefix = "spring.cloud.task.batch.events.skip", name = "enabled", havingValue = "true", matchIfMissing = true)
// @checkstyle:on
public SkipListener skipEventsListener() {
return new EventEmittingSkipListener(this.listenerChannels.skipEvents(),
this.taskEventProperties.getItemProcessOrder());
}
}
}

19
spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingChunkListener.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2017 the original author or authors.
* Copyright 2015-2019 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.
@@ -13,10 +13,12 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.task.batch.listener;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.core.ChunkListener;
import org.springframework.batch.core.scope.context.ChunkContext;
import org.springframework.cloud.task.batch.listener.support.MessagePublisher;
@@ -25,11 +27,12 @@ import org.springframework.messaging.MessageChannel;
import org.springframework.util.Assert;
/**
* Provides informational messages around the {@link org.springframework.batch.core.step.item.Chunk} of a batch job.
* Provides informational messages around the
* {@link org.springframework.batch.core.step.item.Chunk} of a batch job.
*
* The {@link ChunkListener#beforeChunk(ChunkContext)} and
* {@link ChunkListener#afterChunk(ChunkContext)} are both no-ops in this implementation.
* {@link ChunkListener#afterChunkError(ChunkContext)}.
* The {@link ChunkListener#beforeChunk(ChunkContext)} and
* {@link ChunkListener#afterChunk(ChunkContext)} are both no-ops in this implementation.
* {@link ChunkListener#afterChunkError(ChunkContext)}.
*
* @author Ali Shahbour
*/
@@ -38,6 +41,7 @@ public class EventEmittingChunkListener implements ChunkListener, Ordered {
private static final Log logger = LogFactory.getLog(EventEmittingChunkListener.class);
private MessagePublisher<String> messagePublisher;
private int order = Ordered.LOWEST_PRECEDENCE;
public EventEmittingChunkListener(MessageChannel output) {
@@ -52,12 +56,12 @@ public class EventEmittingChunkListener implements ChunkListener, Ordered {
@Override
public void beforeChunk(ChunkContext context) {
messagePublisher.publish("Before Chunk Processing");
this.messagePublisher.publish("Before Chunk Processing");
}
@Override
public void afterChunk(ChunkContext context) {
messagePublisher.publish("After Chunk Processing");
this.messagePublisher.publish("After Chunk Processing");
}
@Override
@@ -69,4 +73,5 @@ public class EventEmittingChunkListener implements ChunkListener, Ordered {
public int getOrder() {
return this.order;
}
}

14
spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingItemProcessListener.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016 the original author or authors.
* Copyright 2015-2019 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.
@@ -13,6 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.task.batch.listener;
import org.springframework.batch.core.ItemProcessListener;
@@ -40,6 +41,7 @@ import org.springframework.util.Assert;
public class EventEmittingItemProcessListener implements ItemProcessListener, Ordered {
private MessagePublisher<String> messagePublisher;
private int order = Ordered.LOWEST_PRECEDENCE;
public EventEmittingItemProcessListener(MessageChannel output) {
@@ -59,23 +61,25 @@ public class EventEmittingItemProcessListener implements ItemProcessListener, Or
@Override
public void afterProcess(Object item, Object result) {
if (result == null) {
messagePublisher.publish("1 item was filtered");
this.messagePublisher.publish("1 item was filtered");
}
else if (item.equals(result)) {
messagePublisher.publish("item equaled result after processing");
this.messagePublisher.publish("item equaled result after processing");
}
else {
messagePublisher.publish("item did not equal result after processing");
this.messagePublisher.publish("item did not equal result after processing");
}
}
@Override
public void onProcessError(Object item, Exception e) {
messagePublisher.publishWithThrowableHeader("Exception while item was being processed", e.getMessage());
this.messagePublisher.publishWithThrowableHeader(
"Exception while item was being processed", e.getMessage());
}
@Override
public int getOrder() {
return this.order;
}
}

38
spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingItemReadListener.java
View File

@@ -1,17 +1,17 @@
/*
* Copyright 2016 the original author or authors.
* Copyright 2015-2019 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
* 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
* 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.
* 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.cloud.task.batch.listener;
@@ -28,21 +28,23 @@ import org.springframework.messaging.MessageChannel;
import org.springframework.util.Assert;
/**
* Provides informational messages around the {@link ItemReader} of a batch job.
* Provides informational messages around the {@link ItemReader} of a batch job.
*
* The {@link ItemReadListener#beforeRead()} and
* {@link ItemReadListener#afterRead(Object)} are both no-ops in this implementation.
* {@link ItemReadListener#onReadError(Exception)} provides the exception
* via the {@link BatchJobHeaders#BATCH_EXCEPTION} message header.
* The {@link ItemReadListener#beforeRead()} and
* {@link ItemReadListener#afterRead(Object)} are both no-ops in this implementation.
* {@link ItemReadListener#onReadError(Exception)} provides the exception via the
* {@link BatchJobHeaders#BATCH_EXCEPTION} message header.
*
* @author Glenn Renfro
* @author Ali Shahbour
*/
public class EventEmittingItemReadListener implements ItemReadListener, Ordered {
private static final Log logger = LogFactory.getLog(EventEmittingItemReadListener.class);
private static final Log logger = LogFactory
.getLog(EventEmittingItemReadListener.class);
private MessagePublisher<String> messagePublisher;
private int order = Ordered.LOWEST_PRECEDENCE;
public EventEmittingItemReadListener(MessageChannel output) {
@@ -71,11 +73,13 @@ public class EventEmittingItemReadListener implements ItemReadListener, Ordered
logger.debug("Executing onReadError: " + ex.getMessage(), ex);
}
messagePublisher.publishWithThrowableHeader("Exception while item was being read", ex.getMessage());
this.messagePublisher.publishWithThrowableHeader(
"Exception while item was being read", ex.getMessage());
}
@Override
public int getOrder() {
return this.order;
}
}

37
spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingItemWriteListener.java
View File

@@ -1,17 +1,17 @@
/*
* Copyright 2016 the original author or authors.
* Copyright 2015-2019 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
* 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
* 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.
* 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.cloud.task.batch.listener;
@@ -29,20 +29,23 @@ import org.springframework.messaging.MessageChannel;
import org.springframework.util.Assert;
/**
* Setups up the ItemWriteEventsListener to emit events to the spring cloud stream output channel.
* Setups up the ItemWriteEventsListener to emit events to the spring cloud stream output
* channel.
*
* Each method provides an informational message.
* {@link ItemWriteListener#onWriteError(Exception, List)} provides a message as well as
* the exception's message via the {@link BatchJobHeaders#BATCH_EXCEPTION} message header.
* Each method provides an informational message.
* {@link ItemWriteListener#onWriteError(Exception, List)} provides a message as well as
* the exception's message via the {@link BatchJobHeaders#BATCH_EXCEPTION} message header.
*
* @author Glenn Renfro
* @author Ali Shahbour
*/
public class EventEmittingItemWriteListener implements ItemWriteListener, Ordered {
private static final Log logger = LogFactory.getLog(EventEmittingItemWriteListener.class);
private static final Log logger = LogFactory
.getLog(EventEmittingItemWriteListener.class);
private MessagePublisher<String> messagePublisher;
private int order = Ordered.LOWEST_PRECEDENCE;
public EventEmittingItemWriteListener(MessageChannel output) {
@@ -73,7 +76,8 @@ public class EventEmittingItemWriteListener implements ItemWriteListener, Ordere
if (logger.isDebugEnabled()) {
logger.debug("Executing onWriteError: " + exception.getMessage(), exception);
}
String payload = "Exception while " + items.size() + " items are attempted to be written.";
String payload = "Exception while " + items.size()
+ " items are attempted to be written.";
this.messagePublisher.publishWithThrowableHeader(payload, exception.getMessage());
}
@@ -81,4 +85,5 @@ public class EventEmittingItemWriteListener implements ItemWriteListener, Ordere
public int getOrder() {
return this.order;
}
}

23
spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingJobExecutionListener.java
View File

@@ -1,18 +1,19 @@
/*
* Copyright 2016 the original author or authors.
* Copyright 2015-2019 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
* 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
* 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.
* 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.cloud.task.batch.listener;
import org.springframework.batch.core.JobExecution;
@@ -33,6 +34,7 @@ import org.springframework.util.Assert;
public class EventEmittingJobExecutionListener implements JobExecutionListener, Ordered {
private MessagePublisher<JobExecutionEvent> messagePublisher;
private int order = Ordered.LOWEST_PRECEDENCE;
public EventEmittingJobExecutionListener(MessageChannel output) {
@@ -59,4 +61,5 @@ public class EventEmittingJobExecutionListener implements JobExecutionListener,
public int getOrder() {
return this.order;
}
}

34
spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingSkipListener.java
View File

@@ -1,17 +1,17 @@
/*
* Copyright 2016 the original author or authors.
* Copyright 2015-2019 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
* 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
* 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.
* 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.cloud.task.batch.listener;
@@ -27,10 +27,11 @@ import org.springframework.messaging.MessageChannel;
import org.springframework.util.Assert;
/**
* Setups up the SkipProcessListener to emit events to the spring cloud stream output channel.
* Setups up the SkipProcessListener to emit events to the spring cloud stream output
* channel.
*
* This listener emits the exception's message via the
* {@link BatchJobHeaders#BATCH_EXCEPTION} message header for each method. For
* {@link BatchJobHeaders#BATCH_EXCEPTION} message header for each method. For
* {@link SkipListener#onSkipInProcess(Object, Throwable)} and
* {@link SkipListener#onSkipInWrite(Object, Throwable)} the body of the message consists
* of the item that caused the error.
@@ -43,6 +44,7 @@ public class EventEmittingSkipListener implements SkipListener, Ordered {
private static final Log logger = LogFactory.getLog(EventEmittingSkipListener.class);
private MessagePublisher<Object> messagePublisher;
private int order = Ordered.LOWEST_PRECEDENCE;
public EventEmittingSkipListener(MessageChannel output) {
@@ -60,7 +62,8 @@ public class EventEmittingSkipListener implements SkipListener, Ordered {
if (logger.isDebugEnabled()) {
logger.debug("Executing onSkipInRead: " + t.getMessage(), t);
}
messagePublisher.publishWithThrowableHeader("Skipped when reading.", t.getMessage());
this.messagePublisher.publishWithThrowableHeader("Skipped when reading.",
t.getMessage());
}
@Override
@@ -68,7 +71,7 @@ public class EventEmittingSkipListener implements SkipListener, Ordered {
if (logger.isDebugEnabled()) {
logger.debug("Executing onSkipInWrite: " + t.getMessage(), t);
}
messagePublisher.publishWithThrowableHeader(item, t.getMessage());
this.messagePublisher.publishWithThrowableHeader(item, t.getMessage());
}
@Override
@@ -76,11 +79,12 @@ public class EventEmittingSkipListener implements SkipListener, Ordered {
if (logger.isDebugEnabled()) {
logger.debug("Executing onSkipInProcess: " + t.getMessage(), t);
}
messagePublisher.publishWithThrowableHeader(item, t.getMessage());
this.messagePublisher.publishWithThrowableHeader(item, t.getMessage());
}
@Override
public int getOrder() {
return this.order;
}
}

12
spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingStepExecutionListener.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016 the original author or authors.
* Copyright 2015-2019 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.
@@ -13,29 +13,32 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.task.batch.listener;
import org.springframework.batch.core.ExitStatus;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.StepExecutionListener;
import org.springframework.cloud.task.batch.listener.support.StepExecutionEvent;
import org.springframework.cloud.task.batch.listener.support.MessagePublisher;
import org.springframework.cloud.task.batch.listener.support.StepExecutionEvent;
import org.springframework.core.Ordered;
import org.springframework.messaging.MessageChannel;
import org.springframework.util.Assert;
/**
* Provides a {@link StepExecutionEvent} at the start and end of each step indicating the
* step's status. The {@link StepExecutionListener#afterStep(StepExecution)} returns the
* step's status. The {@link StepExecutionListener#afterStep(StepExecution)} returns the
* {@link ExitStatus} of the inputted {@link StepExecution}.
*
* @author Michael Minella
* @author Glenn Renfro
* @author Ali Shahbour
*/
public class EventEmittingStepExecutionListener implements StepExecutionListener, Ordered {
public class EventEmittingStepExecutionListener
implements StepExecutionListener, Ordered {
private MessagePublisher<StepExecutionEvent> messagePublisher;
private int order = Ordered.LOWEST_PRECEDENCE;
public EventEmittingStepExecutionListener(MessageChannel output) {
@@ -64,4 +67,5 @@ public class EventEmittingStepExecutionListener implements StepExecutionListener
public int getOrder() {
return this.order;
}
}

26
spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/support/BatchJobHeaders.java
View File

@@ -1,17 +1,17 @@
/*
* Copyright 2016 the original author or authors.
* Copyright 2015-2019 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
* 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
* 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.
* 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.cloud.task.batch.listener.support;
@@ -24,8 +24,14 @@ package org.springframework.cloud.task.batch.listener.support;
*/
public final class BatchJobHeaders {
/**
* Name of the batch listener event type.
*/
public static final String BATCH_LISTENER_EVENT_TYPE = "batch_listener_event_type";
/**
* Name of the batch exception.
*/
public static final String BATCH_EXCEPTION = "batch_exception";
private BatchJobHeaders() {

32
spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/support/ExitStatus.java
View File

@@ -1,17 +1,17 @@
/*
* Copyright 2016 the original author or authors.
* Copyright 2015-2019 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
* 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
* 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.
* 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.cloud.task.batch.listener.support;
@@ -19,8 +19,9 @@ package org.springframework.cloud.task.batch.listener.support;
import org.springframework.util.Assert;
/**
* ExitStatus DTO created so that {@link org.springframework.batch.core.ExitStatus} can be serialized into Json without
* having to add mixins to an ObjectMapper
* ExitStatus DTO created so that {@link org.springframework.batch.core.ExitStatus} can be
* serialized into Json without. having to add mixins to an ObjectMapper
*
* @author Glenn Renfro
*/
public class ExitStatus {
@@ -29,7 +30,7 @@ public class ExitStatus {
private String exitDescription;
public ExitStatus(){
public ExitStatus() {
}
public ExitStatus(org.springframework.batch.core.ExitStatus exitStatus) {
@@ -40,7 +41,7 @@ public class ExitStatus {
}
public String getExitCode() {
return exitCode;
return this.exitCode;
}
public void setExitCode(String exitCode) {
@@ -48,10 +49,11 @@ public class ExitStatus {
}
public String getExitDescription() {
return exitDescription;
return this.exitDescription;
}
public void setExitDescription(String exitDescription) {
this.exitDescription = exitDescription;
}
}

128
spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/support/JobExecutionEvent.java
View File

@@ -1,17 +1,17 @@
/*
* Copyright 2016 the original author or authors.
* Copyright 2015-2019 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
* 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
* 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.
* 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.cloud.task.batch.listener.support;
@@ -33,8 +33,10 @@ import org.springframework.batch.core.StepExecution;
import org.springframework.batch.item.ExecutionContext;
/**
* This is a JobEvent DTO created so that a {@link org.springframework.batch.core.JobExecution} can be serialized into
* Json without having to add mixins to an ObjectMapper.
* This is a JobEvent DTO created so that a
* {@link org.springframework.batch.core.JobExecution} can be serialized into Json without
* having to add mixins to an ObjectMapper.
*
* @author Glenn Renfro
*/
public class JobExecutionEvent extends Entity {
@@ -43,7 +45,7 @@ public class JobExecutionEvent extends Entity {
private JobInstanceEvent jobInstance;
private Collection<StepExecutionEvent> stepExecutions = new CopyOnWriteArraySet<StepExecutionEvent>();
private Collection<StepExecutionEvent> stepExecutions = new CopyOnWriteArraySet<>();
private BatchStatus status = BatchStatus.STARTING;
@@ -55,11 +57,12 @@ public class JobExecutionEvent extends Entity {
private Date lastUpdated = null;
private ExitStatus exitStatus = new ExitStatus(new org.springframework.batch.core.ExitStatus("UNKNOWN"));
private ExitStatus exitStatus = new ExitStatus(
new org.springframework.batch.core.ExitStatus("UNKNOWN"));
private ExecutionContext executionContext = new ExecutionContext();
private List<Throwable> failureExceptions = new CopyOnWriteArrayList<Throwable>();
private List<Throwable> failureExceptions = new CopyOnWriteArrayList<>();
private String jobConfigurationName;
@@ -69,14 +72,15 @@ public class JobExecutionEvent extends Entity {
/**
* Constructor for the StepExecution to initialize the DTO.
*
* @param original the StepExecution to build this DTO around.
*/
public JobExecutionEvent(JobExecution original) {
this.jobParameters = new JobParametersEvent(original.getJobParameters().getParameters());
this.jobInstance = new JobInstanceEvent(original.getJobInstance().getId(), original.getJobInstance().getJobName());
for(StepExecution stepExecution : original.getStepExecutions()){
stepExecutions.add(new StepExecutionEvent(stepExecution));
this.jobParameters = new JobParametersEvent(
original.getJobParameters().getParameters());
this.jobInstance = new JobInstanceEvent(original.getJobInstance().getId(),
original.getJobInstance().getJobName());
for (StepExecution stepExecution : original.getStepExecutions()) {
this.stepExecutions.add(new StepExecutionEvent(stepExecution));
}
this.status = original.getStatus();
this.startTime = original.getStartTime();
@@ -99,16 +103,12 @@ public class JobExecutionEvent extends Entity {
return this.endTime;
}
public void setJobInstance(JobInstanceEvent jobInstance) {
this.jobInstance = jobInstance;
}
public void setEndTime(Date endTime) {
this.endTime = endTime;
}
public Date getStartTime() {
return startTime;
return this.startTime;
}
public void setStartTime(Date startTime) {
@@ -121,7 +121,6 @@ public class JobExecutionEvent extends Entity {
/**
* Set the value of the status field.
*
* @param status the status to set
*/
public void setStatus(BatchStatus status) {
@@ -129,10 +128,9 @@ public class JobExecutionEvent extends Entity {
}
/**
* Upgrade the status field if the provided value is greater than the
* existing one. Clients using this method to set the status can be sure
* that they don't overwrite a failed status with an successful one.
*
* Upgrade the status field if the provided value is greater than the existing one.
* Clients using this method to set the status can be sure that they don't overwrite a
* failed status with an successful one.
* @param status the new status value
*/
public void upgradeStatus(BatchStatus status) {
@@ -142,23 +140,15 @@ public class JobExecutionEvent extends Entity {
/**
* Convenience getter for for the id of the enclosing job. Useful for DAO
* implementations.
*
* @return the id of the enclosing job
*/
public Long getJobId() {
if (jobInstance != null) {
return jobInstance.getId();
if (this.jobInstance != null) {
return this.jobInstance.getId();
}
return null;
}
/**
* @param exitStatus the exit status for the job.
*/
public void setExitStatus(ExitStatus exitStatus) {
this.exitStatus = exitStatus;
}
/**
* @return the exitCode for the job.
*/
@@ -166,6 +156,13 @@ public class JobExecutionEvent extends Entity {
return this.exitStatus;
}
/**
* @param exitStatus the exit status for the job.
*/
public void setExitStatus(ExitStatus exitStatus) {
this.exitStatus = exitStatus;
}
/**
* @return the Job that is executing.
*/
@@ -173,9 +170,12 @@ public class JobExecutionEvent extends Entity {
return this.jobInstance;
}
public void setJobInstance(JobInstanceEvent jobInstance) {
this.jobInstance = jobInstance;
}
/**
* Accessor for the step executions.
*
* @return the step executions that were registered
*/
public Collection<StepExecutionEvent> getStepExecutions() {
@@ -183,24 +183,22 @@ public class JobExecutionEvent extends Entity {
}
/**
* Sets the {@link ExecutionContext} for this execution
*
* @param executionContext the context
*/
public void setExecutionContext(ExecutionContext executionContext) {
this.executionContext = executionContext;
}
/**
* Returns the {@link ExecutionContext} for this execution. The content is
* expected to be persisted after each step completion (successful or not).
*
* Returns the {@link ExecutionContext} for this execution. The content is expected to
* be persisted after each step completion (successful or not).
* @return the context
*/
public ExecutionContext getExecutionContext() {
return this.executionContext;
}
/**
* Sets the {@link ExecutionContext} for this execution.
* @param executionContext the context
*/
public void setExecutionContext(ExecutionContext executionContext) {
this.executionContext = executionContext;
}
/**
* @return the time when this execution was created.
*/
@@ -220,9 +218,8 @@ public class JobExecutionEvent extends Entity {
}
/**
* Get the date representing the last time this JobExecution was updated in
* the JobRepository.
*
* Get the date representing the last time this JobExecution was updated in the
* JobRepository.
* @return Date representing the last time this JobExecution was updated.
*/
public Date getLastUpdated() {
@@ -231,7 +228,6 @@ public class JobExecutionEvent extends Entity {
/**
* Set the last time this {@link JobExecution} was updated.
*
* @param lastUpdated The date the {@link JobExecution} was updated.
*/
public void setLastUpdated(Date lastUpdated) {
@@ -244,7 +240,6 @@ public class JobExecutionEvent extends Entity {
/**
* Add the provided throwable to the failure exception list.
*
* @param t a {@link Throwable} to be added to the exception list.
*/
public synchronized void addFailureException(Throwable t) {
@@ -252,11 +247,10 @@ public class JobExecutionEvent extends Entity {
}
/**
* Return all failure causing exceptions for this JobExecution, including
* step executions.
*
* @return List&lt;Throwable&gt; containing all exceptions causing failure for
* this JobExecution.
* Return all failure causing exceptions for this JobExecution, including step
* executions.
* @return List&lt;Throwable&gt; containing all exceptions causing failure for this
* JobExecution.
*/
public synchronized List<Throwable> getAllFailureExceptions() {
@@ -275,8 +269,10 @@ public class JobExecutionEvent extends Entity {
*/
@Override
public String toString() {
return super.toString()
+ String.format(", startTime=%s, endTime=%s, lastUpdated=%s, status=%s, exitStatus=%s, job=[%s], jobParameters=[%s]",
startTime, endTime, lastUpdated, status, exitStatus, jobInstance, jobParameters);
return super.toString() + String.format(
", startTime=%s, endTime=%s, lastUpdated=%s, status=%s, exitStatus=%s, job=[%s], jobParameters=[%s]",
this.startTime, this.endTime, this.lastUpdated, this.status,
this.exitStatus, this.jobInstance, this.jobParameters);
}
}

29
spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/support/JobInstanceEvent.java
View File

@@ -1,17 +1,17 @@
/*
* Copyright 2016 the original author or authors.
* Copyright 2015-2019 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
* 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
* 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.
* 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.cloud.task.batch.listener.support;
@@ -20,8 +20,9 @@ import org.springframework.batch.core.Entity;
import org.springframework.util.Assert;
/**
* This is a JobInstance DTO created so that a {@link org.springframework.batch.core.JobInstance} can be serialized into
* Json without having to add mixins to an ObjectMapper.
* This is a JobInstance DTO created so that a
* {@link org.springframework.batch.core.JobInstance} can be serialized into Json without
* having to add mixins to an ObjectMapper.
*
* @author Glenn Renfro
*/
@@ -44,11 +45,11 @@ public class JobInstanceEvent extends Entity {
* @return the job name. (Equivalent to getJob().getName())
*/
public String getJobName() {
return jobName;
return this.jobName;
}
public String toString() {
return super.toString() + ", Job=[" + jobName + "]";
return super.toString() + ", Job=[" + this.jobName + "]";
}
public long getInstanceId() {

63
spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/support/JobParameterEvent.java
View File

@@ -1,17 +1,17 @@
/*
* Copyright 2016 the original author or authors.
* Copyright 2015-2019 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
* 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
* 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.
* 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.cloud.task.batch.listener.support;
@@ -21,11 +21,14 @@ import java.util.Date;
import org.springframework.batch.core.JobParameter;
/**
* This is a JobParameter DTO created so that a {@link org.springframework.batch.core.JobParameter} can be serialized
* into Json without having to add mixins to an ObjectMapper.
* This is a JobParameter DTO created so that a
* {@link org.springframework.batch.core.JobParameter} can be serialized into Json without
* having to add mixins to an ObjectMapper.
*
* @author Glenn Renfro
*/
public class JobParameterEvent {
private Object parameter;
private JobParameterEvent.ParameterType parameterType;
@@ -42,7 +45,7 @@ public class JobParameterEvent {
}
public boolean isIdentifying() {
return identifying;
return this.identifying;
}
/**
@@ -50,11 +53,11 @@ public class JobParameterEvent {
*/
public Object getValue() {
if (parameter != null && parameter.getClass().isInstance(Date.class)) {
return new Date(((Date) parameter).getTime());
if (this.parameter != null && this.parameter.getClass().isInstance(Date.class)) {
return new Date(((Date) this.parameter).getTime());
}
else {
return parameter;
return this.parameter;
}
}
@@ -62,7 +65,7 @@ public class JobParameterEvent {
* @return a ParameterType representing the type of this parameter.
*/
public JobParameterEvent.ParameterType getType() {
return parameterType;
return this.parameterType;
}
@Override
@@ -76,44 +79,54 @@ public class JobParameterEvent {
}
JobParameterEvent rhs = (JobParameterEvent) obj;
return parameter==null ? rhs.parameter==null && parameterType==rhs.parameterType: parameter.equals(rhs.parameter);
return this.parameter == null
? rhs.parameter == null && this.parameterType == rhs.parameterType
: this.parameter.equals(rhs.parameter);
}
@Override
public String toString() {
return parameter == null ? null : (parameterType == JobParameterEvent.ParameterType.DATE ? "" + ((Date) parameter).getTime()
: parameter.toString());
return this.parameter == null ? null
: (this.parameterType == JobParameterEvent.ParameterType.DATE
? "" + ((Date) this.parameter).getTime()
: this.parameter.toString());
}
@Override
public int hashCode() {
final int BASE_HASH = 7;
final int MULTIPLIER_HASH = 21;
return BASE_HASH + MULTIPLIER_HASH * (parameter == null ? parameterType.hashCode() : parameter.hashCode());
return BASE_HASH + MULTIPLIER_HASH * (this.parameter == null
? this.parameterType.hashCode() : this.parameter.hashCode());
}
/**
* Enumeration representing the type of a JobParameter.
*/
public enum ParameterType {
// @checkstyle:off
STRING, DATE, LONG, DOUBLE;
// @checkstyle:on
public static ParameterType convert(JobParameter.ParameterType type) {
if(JobParameter.ParameterType.DATE.equals(type)) {
if (JobParameter.ParameterType.DATE.equals(type)) {
return DATE;
}
else if(JobParameter.ParameterType.DOUBLE.equals(type)) {
else if (JobParameter.ParameterType.DOUBLE.equals(type)) {
return DOUBLE;
}
else if(JobParameter.ParameterType.LONG.equals(type)) {
else if (JobParameter.ParameterType.LONG.equals(type)) {
return LONG;
}
else if(JobParameter.ParameterType.STRING.equals(type)) {
else if (JobParameter.ParameterType.STRING.equals(type)) {
return STRING;
}
else {
throw new IllegalArgumentException("Unable to convert type");
}
}
}
}

165
spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/support/JobParametersEvent.java
View File

@@ -1,17 +1,17 @@
/*
* Copyright 2016 the original author or authors.
* Copyright 2015-2019 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
* 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
* 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.
* 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.cloud.task.batch.listener.support;
@@ -24,185 +24,177 @@ import java.util.Properties;
import org.springframework.batch.core.JobParameter;
/**
* This is a JobParametersEvent DTO created so that a {@link org.springframework.batch.core.JobParameters} can be
* serialized into Json without having to add mixins to an ObjectMapper.
* This is a JobParametersEvent DTO created so that a
* {@link org.springframework.batch.core.JobParameters} can be serialized into Json
* without having to add mixins to an ObjectMapper.
*
* @author Glenn Renfro
*/
public class JobParametersEvent {
private final Map<String,JobParameterEvent> parameters;
private final Map<String, JobParameterEvent> parameters;
public JobParametersEvent() {
this.parameters = new LinkedHashMap<>();
}
public JobParametersEvent(Map<String,JobParameter> jobParameters) {
public JobParametersEvent(Map<String, JobParameter> jobParameters) {
this.parameters = new LinkedHashMap<>();
for(Map.Entry<String, JobParameter> entry: jobParameters.entrySet()){
if(entry.getValue().getValue() instanceof String){
parameters.put(entry.getKey(), new JobParameterEvent(entry.getValue()));
for (Map.Entry<String, JobParameter> entry : jobParameters.entrySet()) {
if (entry.getValue().getValue() instanceof String) {
this.parameters.put(entry.getKey(),
new JobParameterEvent(entry.getValue()));
}
else if(entry.getValue().getValue() instanceof Long){
parameters.put(entry.getKey(), new JobParameterEvent(entry.getValue()));
else if (entry.getValue().getValue() instanceof Long) {
this.parameters.put(entry.getKey(),
new JobParameterEvent(entry.getValue()));
}
else if(entry.getValue().getValue() instanceof Date){
parameters.put(entry.getKey(), new JobParameterEvent(entry.getValue()));
else if (entry.getValue().getValue() instanceof Date) {
this.parameters.put(entry.getKey(),
new JobParameterEvent(entry.getValue()));
}
else if(entry.getValue().getValue() instanceof Double){
parameters.put(entry.getKey(), new JobParameterEvent(entry.getValue()));
else if (entry.getValue().getValue() instanceof Double) {
this.parameters.put(entry.getKey(),
new JobParameterEvent(entry.getValue()));
}
}
}
/**
* Typesafe Getter for the Long represented by the provided key.
*
* @param key The key to get a value for
* @return The <code>Long</code> value
*/
public Long getLong(String key){
if (!parameters.containsKey(key)) {
public Long getLong(String key) {
if (!this.parameters.containsKey(key)) {
return 0L;
}
Object value = parameters.get(key).getValue();
return value==null ? 0L : ((Long)value).longValue();
Object value = this.parameters.get(key).getValue();
return value == null ? 0L : ((Long) value).longValue();
}
/**
* Typesafe Getter for the Long represented by the provided key. If the
* key does not exist, the default value will be returned.
*
* Typesafe Getter for the Long represented by the provided key. If the key does not
* exist, the default value will be returned.
* @param key to return the value for
* @param defaultValue to return if the value doesn't exist
* @return the parameter represented by the provided key, defaultValue
* otherwise.
* @return the parameter represented by the provided key, defaultValue otherwise.
*/
public Long getLong(String key, long defaultValue){
if(parameters.containsKey(key)){
public Long getLong(String key, long defaultValue) {
if (this.parameters.containsKey(key)) {
return getLong(key);
}
else{
else {
return defaultValue;
}
}
/**
* Typesafe Getter for the String represented by the provided key.
*
* @param key The key to get a value for
* @return The <code>String</code> value
*/
public String getString(String key){
JobParameterEvent value = parameters.get(key);
return value==null ? null : value.toString();
public String getString(String key) {
JobParameterEvent value = this.parameters.get(key);
return value == null ? null : value.toString();
}
/**
* Typesafe Getter for the String represented by the provided key. If the
* key does not exist, the default value will be returned.
*
* Typesafe Getter for the String represented by the provided key. If the key does not
* exist, the default value will be returned.
* @param key to return the value for
* @param defaultValue to return if the value doesn't exist
* @return the parameter represented by the provided key, defaultValue
* otherwise.
* @return the parameter represented by the provided key, defaultValue otherwise.
*/
public String getString(String key, String defaultValue){
if(parameters.containsKey(key)){
public String getString(String key, String defaultValue) {
if (this.parameters.containsKey(key)) {
return getString(key);
}
else{
else {
return defaultValue;
}
}
/**
* Typesafe Getter for the Long represented by the provided key.
*
* @param key The key to get a value for
* @return The <code>Double</code> value
*/
public Double getDouble(String key){
if (!parameters.containsKey(key)) {
public Double getDouble(String key) {
if (!this.parameters.containsKey(key)) {
return 0.0;
}
Double value = (Double)parameters.get(key).getValue();
return value==null ? 0.0 : value.doubleValue();
Double value = (Double) this.parameters.get(key).getValue();
return value == null ? 0.0 : value.doubleValue();
}
/**
* Typesafe Getter for the Double represented by the provided key. If the
* key does not exist, the default value will be returned.
*
* Typesafe Getter for the Double represented by the provided key. If the key does not
* exist, the default value will be returned.
* @param key to return the value for
* @param defaultValue to return if the value doesn't exist
* @return the parameter represented by the provided key, defaultValue
* otherwise.
* @return the parameter represented by the provided key, defaultValue otherwise.
*/
public Double getDouble(String key, double defaultValue){
if(parameters.containsKey(key)){
public Double getDouble(String key, double defaultValue) {
if (this.parameters.containsKey(key)) {
return getDouble(key);
}
else{
else {
return defaultValue;
}
}
/**
* Typesafe Getter for the Date represented by the provided key.
*
* @param key The key to get a value for
* @return The <code>java.util.Date</code> value
*/
public Date getDate(String key){
return this.getDate(key,null);
public Date getDate(String key) {
return this.getDate(key, null);
}
/**
* Typesafe Getter for the Date represented by the provided key. If the
* key does not exist, the default value will be returned.
*
* Typesafe Getter for the Date represented by the provided key. If the key does not
* exist, the default value will be returned.
* @param key to return the value for
* @param defaultValue to return if the value doesn't exist
* @return the parameter represented by the provided key, defaultValue
* otherwise.
* @return the parameter represented by the provided key, defaultValue otherwise.
*/
public Date getDate(String key, Date defaultValue){
if(parameters.containsKey(key)){
return (Date)parameters.get(key).getValue();
public Date getDate(String key, Date defaultValue) {
if (this.parameters.containsKey(key)) {
return (Date) this.parameters.get(key).getValue();
}
else{
else {
return defaultValue;
}
}
/**
* Get a map of all parameters, including string, long, and date.
*
* @return an unmodifiable map containing all parameters.
*/
public Map<String, JobParameterEvent> getParameters(){
return new LinkedHashMap<String, JobParameterEvent>(parameters);
public Map<String, JobParameterEvent> getParameters() {
return new LinkedHashMap<>(this.parameters);
}
/**
* @return true if the parameters is empty, false otherwise.
*/
public boolean isEmpty(){
return parameters.isEmpty();
public boolean isEmpty() {
return this.parameters.isEmpty();
}
@Override
public boolean equals(Object obj) {
if(obj instanceof JobParametersEvent == false){
if (!(obj instanceof JobParametersEvent)) {
return false;
}
if(obj == this){
if (obj == this) {
return true;
}
JobParametersEvent rhs = (JobParametersEvent)obj;
JobParametersEvent rhs = (JobParametersEvent) obj;
return this.parameters.equals(rhs.parameters);
}
@@ -210,23 +202,24 @@ public class JobParametersEvent {
public int hashCode() {
final int BASE_HASH = 17;
final int MULTIPLIER_HASH = 23;
return BASE_HASH + MULTIPLIER_HASH * parameters.hashCode();
return BASE_HASH + MULTIPLIER_HASH * this.parameters.hashCode();
}
@Override
public String toString() {
return parameters.toString();
return this.parameters.toString();
}
public Properties toProperties() {
Properties props = new Properties();
for (Map.Entry<String, JobParameterEvent> param : parameters.entrySet()) {
if(param.getValue() != null) {
for (Map.Entry<String, JobParameterEvent> param : this.parameters.entrySet()) {
if (param.getValue() != null) {
props.put(param.getKey(), param.getValue().toString());
}
}
return props;
}
}

32
spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/support/MessagePublisher.java
View File

@@ -1,17 +1,17 @@
/*
* Copyright 2016 the original author or authors.
* Copyright 2015-2019 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
* 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
* 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.
* 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.cloud.task.batch.listener.support;
@@ -22,10 +22,13 @@ import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.Assert;
/**
* Utility class that sends batch job listener payloads to the notification channel.
* Utility class that sends batch job listener payloads to the notification channel.
*
* @param <P> payload type
* @author Glenn Renfro
*/
public class MessagePublisher<P>{
public class MessagePublisher<P> {
private final MessageChannel listenerEventsChannel;
public MessagePublisher(MessageChannel listenerEventsChannel) {
@@ -48,8 +51,9 @@ public class MessagePublisher<P>{
}
public void publishWithThrowableHeader(P payload, String header) {
Message<P> message = MessageBuilder.withPayload(payload).setHeader(BatchJobHeaders.BATCH_EXCEPTION,
header).build();
Message<P> message = MessageBuilder.withPayload(payload)
.setHeader(BatchJobHeaders.BATCH_EXCEPTION, header).build();
publishMessage(message);
}
}

191
spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/support/StepExecutionEvent.java
View File

@@ -1,17 +1,17 @@
/*
* Copyright 2016 the original author or authors.
* Copyright 2015-2019 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
* 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
* 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.
* 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.cloud.task.batch.listener.support;
@@ -27,8 +27,10 @@ import org.springframework.batch.item.ExecutionContext;
import org.springframework.util.Assert;
/**
* This is a StepExecution DTO created so that a {@link org.springframework.batch.core.StepExecution} can be serialized
* into Json without having to add mixins to an ObjectMapper.
* This is a StepExecution DTO created so that a
* {@link org.springframework.batch.core.StepExecution} can be serialized into Json
* without having to add mixins to an ObjectMapper.
*
* @author Glenn Renfro
*/
public class StepExecutionEvent extends Entity {
@@ -61,14 +63,14 @@ public class StepExecutionEvent extends Entity {
private ExecutionContext executionContext = new ExecutionContext();
private ExitStatus exitStatus = new ExitStatus(org.springframework.batch.core.ExitStatus.EXECUTING);
private ExitStatus exitStatus = new ExitStatus(
org.springframework.batch.core.ExitStatus.EXECUTING);
private boolean terminateOnly;
private int filterCount;
private List<Throwable> failureExceptions = new CopyOnWriteArrayList<Throwable>();
private List<Throwable> failureExceptions = new CopyOnWriteArrayList<>();
public StepExecutionEvent() {
super();
@@ -76,13 +78,14 @@ public class StepExecutionEvent extends Entity {
/**
* Constructor for the StepExecution to initialize the DTO.
*
* @param stepExecution the StepExecution to build this DTO around.
*/
public StepExecutionEvent(StepExecution stepExecution) {
super();
Assert.notNull(stepExecution, "StepExecution must be provided to re-hydrate an existing StepExecutionEvent");
Assert.notNull(stepExecution.getJobExecution(), "JobExecution must be provided to re-hydrate an existing StepExecutionEvent");
Assert.notNull(stepExecution,
"StepExecution must be provided to re-hydrate an existing StepExecutionEvent");
Assert.notNull(stepExecution.getJobExecution(),
"JobExecution must be provided to re-hydrate an existing StepExecutionEvent");
setId(stepExecution.getId());
this.jobExecutionId = stepExecution.getJobExecutionId();
this.stepName = stepExecution.getStepName();
@@ -90,7 +93,7 @@ public class StepExecutionEvent extends Entity {
this.status = stepExecution.getStatus();
this.exitStatus = new ExitStatus(stepExecution.getExitStatus());
this.executionContext = stepExecution.getExecutionContext();
for (Throwable throwable : stepExecution.getFailureExceptions()){
for (Throwable throwable : stepExecution.getFailureExceptions()) {
this.failureExceptions.add(throwable);
}
this.terminateOnly = stepExecution.isTerminateOnly();
@@ -111,8 +114,7 @@ public class StepExecutionEvent extends Entity {
}
/**
* Returns the {@link ExecutionContext} for this execution
*
* Returns the {@link ExecutionContext} for this execution.
* @return the attributes
*/
public ExecutionContext getExecutionContext() {
@@ -120,8 +122,7 @@ public class StepExecutionEvent extends Entity {
}
/**
* Sets the {@link ExecutionContext} for this execution
*
* Sets the {@link ExecutionContext} for this execution.
* @param executionContext the attributes
*/
public void setExecutionContext(ExecutionContext executionContext) {
@@ -129,8 +130,7 @@ public class StepExecutionEvent extends Entity {
}
/**
* Returns the current number of commits for this execution
*
* Returns the current number of commits for this execution.
* @return the current number of commits
*/
public int getCommitCount() {
@@ -138,8 +138,7 @@ public class StepExecutionEvent extends Entity {
}
/**
* Sets the current number of commits for this execution
*
* Sets the current number of commits for this execution.
* @param commitCount the current number of commits
*/
public void setCommitCount(int commitCount) {
@@ -147,8 +146,7 @@ public class StepExecutionEvent extends Entity {
}
/**
* Returns the time that this execution ended
*
* Returns the time that this execution ended.
* @return the time that this execution ended
*/
public Date getEndTime() {
@@ -156,8 +154,7 @@ public class StepExecutionEvent extends Entity {
}
/**
* Sets the time that this execution ended
*
* Sets the time that this execution ended.
* @param endTime the time that this execution ended
*/
public void setEndTime(Date endTime) {
@@ -165,8 +162,7 @@ public class StepExecutionEvent extends Entity {
}
/**
* Returns the current number of items read for this execution
*
* Returns the current number of items read for this execution.
* @return the current number of items read for this execution
*/
public int getReadCount() {
@@ -174,8 +170,7 @@ public class StepExecutionEvent extends Entity {
}
/**
* Sets the current number of read items for this execution
*
* Sets the current number of read items for this execution.
* @param readCount the current number of read items for this execution
*/
public void setReadCount(int readCount) {
@@ -183,8 +178,7 @@ public class StepExecutionEvent extends Entity {
}
/**
* Returns the current number of items written for this execution
*
* Returns the current number of items written for this execution.
* @return the current number of items written for this execution
*/
public int getWriteCount() {
@@ -192,8 +186,7 @@ public class StepExecutionEvent extends Entity {
}
/**
* Sets the current number of written items for this execution
*
* Sets the current number of written items for this execution.
* @param writeCount the current number of written items for this execution
*/
public void setWriteCount(int writeCount) {
@@ -201,8 +194,7 @@ public class StepExecutionEvent extends Entity {
}
/**
* Returns the current number of rollbacks for this execution
*
* Returns the current number of rollbacks for this execution.
* @return the current number of rollbacks for this execution
*/
public int getRollbackCount() {
@@ -210,8 +202,15 @@ public class StepExecutionEvent extends Entity {
}
/**
* Returns the current number of items filtered out of this execution
*
* Setter for number of rollbacks for this execution.
* @param rollbackCount the number of rollbacks for this execution
*/
public void setRollbackCount(int rollbackCount) {
this.rollbackCount = rollbackCount;
}
/**
* Returns the current number of items filtered out of this execution.
* @return the current number of items filtered out of this execution
*/
public int getFilterCount() {
@@ -220,24 +219,14 @@ public class StepExecutionEvent extends Entity {
/**
* Public setter for the number of items filtered out of this execution.
* @param filterCount the number of items filtered out of this execution to
* set
* @param filterCount the number of items filtered out of this execution to set
*/
public void setFilterCount(int filterCount) {
this.filterCount = filterCount;
}
/**
* Setter for number of rollbacks for this execution
* @param rollbackCount the number of rollbacks for this execution
*/
public void setRollbackCount(int rollbackCount) {
this.rollbackCount = rollbackCount;
}
/**
* Gets the time this execution started
*
* Gets the time this execution started.
* @return the time this execution started
*/
public Date getStartTime() {
@@ -245,8 +234,7 @@ public class StepExecutionEvent extends Entity {
}
/**
* Sets the time this execution started
*
* Sets the time this execution started.
* @param startTime the time this execution started
*/
public void setStartTime(Date startTime) {
@@ -254,8 +242,7 @@ public class StepExecutionEvent extends Entity {
}
/**
* Returns the current status of this step
*
* Returns the current status of this step.
* @return the current status of this step
*/
public BatchStatus getStatus() {
@@ -263,29 +250,22 @@ public class StepExecutionEvent extends Entity {
}
/**
* Sets the current status of this step
*
* Sets the current status of this step.
* @param status the current status of this step
*/
public void setStatus(BatchStatus status) {
this.status = status;
}
public void setStepName(String stepName) {
this.stepName = stepName;
}
/**
* @return the name of the step
* @return the name of the step.
*/
public String getStepName() {
return this.stepName;
}
/**
* @param exitStatus the {@link ExitStatus} for the step.
*/
public void setExitStatus(ExitStatus exitStatus) {
this.exitStatus = exitStatus;
public void setStepName(String stepName) {
this.stepName = stepName;
}
/**
@@ -295,6 +275,13 @@ public class StepExecutionEvent extends Entity {
return this.exitStatus;
}
/**
* @param exitStatus the {@link ExitStatus} for the step.
*/
public void setExitStatus(ExitStatus exitStatus) {
this.exitStatus = exitStatus;
}
/**
* @return flag to indicate that an execution should halt
*/
@@ -303,8 +290,8 @@ public class StepExecutionEvent extends Entity {
}
/**
* Set a flag that will signal to an execution environment that this
* execution (and its surrounding job) wishes to exit.
* Set a flag that will signal to an execution environment that this execution (and
* its surrounding job) wishes to exit.
*/
public void setTerminateOnly() {
this.terminateOnly = true;
@@ -318,19 +305,27 @@ public class StepExecutionEvent extends Entity {
}
/**
* Increment the number of commits
* Increment the number of commits.
*/
public void incrementCommitCount() {
this.commitCount++;
}
/**
* @return the number of records skipped on read
* @return the number of records skipped on read.
*/
public int getReadSkipCount() {
return this.readSkipCount;
}
/**
* Set the number of records skipped on read.
* @param readSkipCount the number of records to be skipped on read.
*/
public void setReadSkipCount(int readSkipCount) {
this.readSkipCount = readSkipCount;
}
/**
* @return the number of records skipped on write
*/
@@ -339,17 +334,7 @@ public class StepExecutionEvent extends Entity {
}
/**
* Set the number of records skipped on read
*
* @param readSkipCount the number of records to be skipped on read.
*/
public void setReadSkipCount(int readSkipCount) {
this.readSkipCount = readSkipCount;
}
/**
* Set the number of records skipped on write
*
* Set the number of records skipped on write.
* @param writeSkipCount the number of records to be skipped on write.
*/
public void setWriteSkipCount(int writeSkipCount) {
@@ -365,7 +350,6 @@ public class StepExecutionEvent extends Entity {
/**
* Set the number of records skipped during processing.
*
* @param processSkipCount the number of records skip during processing.
*/
public void setProcessSkipCount(int processSkipCount) {
@@ -380,8 +364,7 @@ public class StepExecutionEvent extends Entity {
}
/**
* Set the time when the StepExecution was last updated before persisting
*
* Set the time when the StepExecution was last updated before persisting.
* @param lastUpdated the {@link Date} the StepExecution was last updated.
*/
public void setLastUpdated(Date lastUpdated) {
@@ -399,19 +382,19 @@ public class StepExecutionEvent extends Entity {
/*
* (non-Javadoc)
*
* @see
* org.springframework.batch.container.common.domain.Entity#equals(java.
* @see org.springframework.batch.container.common.domain.Entity#equals(java.
* lang.Object)
*/
@Override
public boolean equals(Object obj) {
if ( !(obj instanceof StepExecution) || getId() == null) {
if (!(obj instanceof StepExecution) || getId() == null) {
return super.equals(obj);
}
StepExecution other = (StepExecution) obj;
return this.stepName.equals(other.getStepName()) && (this.jobExecutionId == other.getJobExecutionId())
return this.stepName.equals(other.getStepName())
&& (this.jobExecutionId == other.getJobExecutionId())
&& getId().equals(other.getId());
}
@@ -424,21 +407,27 @@ public class StepExecutionEvent extends Entity {
public int hashCode() {
Object jobExecutionId = getJobExecutionId();
Long id = getId();
return super.hashCode() + 31 * (this.stepName != null ? this.stepName.hashCode() : 0) + 91
* (jobExecutionId != null ? jobExecutionId.hashCode() : 0) + 59 * (id != null ? id.hashCode() : 0);
return super.hashCode()
+ 31 * (this.stepName != null ? this.stepName.hashCode() : 0)
+ 91 * (jobExecutionId != null ? jobExecutionId.hashCode() : 0)
+ 59 * (id != null ? id.hashCode() : 0);
}
@Override
public String toString() {
return String.format(getSummary() + ", exitDescription=%s", this.exitStatus.getExitDescription());
return String.format(getSummary() + ", exitDescription=%s",
this.exitStatus.getExitDescription());
}
public String getSummary() {
return super.toString()
+ String.format(
", name=%s, status=%s, exitStatus=%s, readCount=%d, filterCount=%d, writeCount=%d readSkipCount=%d, writeSkipCount=%d"
+ ", processSkipCount=%d, commitCount=%d, rollbackCount=%d", this.stepName, this.status,
this.exitStatus.getExitCode(), this.readCount, this.filterCount, this.writeCount, this.readSkipCount, this.writeSkipCount,
this.processSkipCount, this.commitCount, this.rollbackCount);
return super.toString() + String.format(
", name=%s, status=%s, exitStatus=%s, readCount=%d, "
+ "filterCount=%d, writeCount=%d readSkipCount=%d, writeSkipCount=%d"
+ ", processSkipCount=%d, commitCount=%d, rollbackCount=%d",
this.stepName, this.status, this.exitStatus.getExitCode(), this.readCount,
this.filterCount, this.writeCount, this.readSkipCount,
this.writeSkipCount, this.processSkipCount, this.commitCount,
this.rollbackCount);
}
}

117
spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/support/TaskBatchEventListenerBeanPostProcessor.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016 the original author or authors.
* Copyright 2015-2019 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.
@@ -13,6 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.task.batch.listener.support;
import java.lang.reflect.Field;
@@ -39,22 +40,27 @@ import org.springframework.context.ApplicationContext;
import org.springframework.util.ReflectionUtils;
/**
* Attaches the listeners to the job and its steps.
* Based on the type of bean that is being processed will determine what listener is attached.
* <ul>
* <li>If the bean is of type AbstactJob then the JobExecutionListener is registered with this bean.</li>
* <li>If the bean is of type AbstactStep then the StepExecutionListener is registered with this bean.</li>
* <li>If the bean is of type TaskletStep then the ChunkEventListener is registered with this bean.</li>
* <li>If the tasklet for the TaskletStep is of type ChunkOrientedTasklet the following listeners will be registered. </li>
* <li>
* <ul>
* <li>ItemReadListener with the ChunkProvider.</li>
* <li>ItemProcessListener with the ChunkProcessor.</li>
* <li>ItemWriteEventsListener with the ChunkProcessor.</li>
* <li>SkipEventsListener with the ChunkProcessor.</li>
* </ul>
* </li>
* </ul>
* Attaches the listeners to the job and its steps. Based on the type of bean that is
* being processed will determine what listener is attached.
* <ul>
* <li>If the bean is of type AbstactJob then the JobExecutionListener is registered with
* this bean.</li>
* <li>If the bean is of type AbstactStep then the StepExecutionListener is registered
* with this bean.</li>
* <li>If the bean is of type TaskletStep then the ChunkEventListener is registered with
* this bean.</li>
* <li>If the tasklet for the TaskletStep is of type ChunkOrientedTasklet the following
* listeners will be registered.</li>
* <li>
* <ul>
* <li>ItemReadListener with the ChunkProvider.</li>
* <li>ItemProcessListener with the ChunkProcessor.</li>
* <li>ItemWriteEventsListener with the ChunkProcessor.</li>
* <li>SkipEventsListener with the ChunkProcessor.</li>
* </ul>
* </li>
* </ul>
*
* @author Michael Minella
* @author Glenn Renfro
*/
@@ -64,7 +70,8 @@ public class TaskBatchEventListenerBeanPostProcessor implements BeanPostProcesso
private ApplicationContext applicationContext;
@Override
public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
public Object postProcessBeforeInitialization(Object bean, String beanName)
throws BeansException {
registerJobExecutionEventListener(bean);
@@ -76,12 +83,16 @@ public class TaskBatchEventListenerBeanPostProcessor implements BeanPostProcesso
registerChunkEventsListener(bean);
if (tasklet instanceof ChunkOrientedTasklet) {
Field chunkProviderField = ReflectionUtils.findField(ChunkOrientedTasklet.class, "chunkProvider");
Field chunkProviderField = ReflectionUtils
.findField(ChunkOrientedTasklet.class, "chunkProvider");
ReflectionUtils.makeAccessible(chunkProviderField);
SimpleChunkProvider chunkProvider = (SimpleChunkProvider) ReflectionUtils.getField(chunkProviderField, tasklet);
Field chunkProcessorField = ReflectionUtils.findField(ChunkOrientedTasklet.class, "chunkProcessor");
SimpleChunkProvider chunkProvider = (SimpleChunkProvider) ReflectionUtils
.getField(chunkProviderField, tasklet);
Field chunkProcessorField = ReflectionUtils
.findField(ChunkOrientedTasklet.class, "chunkProcessor");
ReflectionUtils.makeAccessible(chunkProcessorField);
SimpleChunkProcessor chunkProcessor = (SimpleChunkProcessor) ReflectionUtils.getField(chunkProcessorField, tasklet);
SimpleChunkProcessor chunkProcessor = (SimpleChunkProcessor) ReflectionUtils
.getField(chunkProcessorField, tasklet);
registerItemReadEvents(chunkProvider);
registerSkipEvents(chunkProvider);
registerItemProcessEvents(chunkProcessor);
@@ -95,67 +106,79 @@ public class TaskBatchEventListenerBeanPostProcessor implements BeanPostProcesso
}
@Override
public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
public Object postProcessAfterInitialization(Object bean, String beanName)
throws BeansException {
return bean;
}
private void registerItemProcessEvents(SimpleChunkProcessor chunkProcessor) {
if(this.applicationContext.containsBean(BatchEventAutoConfiguration.ITEM_PROCESS_EVENTS_LISTENER)) {
chunkProcessor.registerListener((ItemProcessListener) this.applicationContext.getBean(BatchEventAutoConfiguration.ITEM_PROCESS_EVENTS_LISTENER));
if (this.applicationContext
.containsBean(BatchEventAutoConfiguration.ITEM_PROCESS_EVENTS_LISTENER)) {
chunkProcessor.registerListener((ItemProcessListener) this.applicationContext
.getBean(BatchEventAutoConfiguration.ITEM_PROCESS_EVENTS_LISTENER));
}
}
private void registerItemReadEvents(SimpleChunkProvider chunkProvider) {
if(this.applicationContext.containsBean(BatchEventAutoConfiguration.ITEM_READ_EVENTS_LISTENER)) {
chunkProvider.registerListener((ItemReadListener) this.applicationContext.getBean(BatchEventAutoConfiguration.ITEM_READ_EVENTS_LISTENER));
if (this.applicationContext
.containsBean(BatchEventAutoConfiguration.ITEM_READ_EVENTS_LISTENER)) {
chunkProvider.registerListener((ItemReadListener) this.applicationContext
.getBean(BatchEventAutoConfiguration.ITEM_READ_EVENTS_LISTENER));
}
}
private void registerItemWriteEvents(SimpleChunkProcessor chunkProcessor) {
if(this.applicationContext.containsBean(BatchEventAutoConfiguration.ITEM_WRITE_EVENTS_LISTENER)) {
chunkProcessor.registerListener((ItemWriteListener) this.applicationContext.getBean(BatchEventAutoConfiguration.ITEM_WRITE_EVENTS_LISTENER));
if (this.applicationContext
.containsBean(BatchEventAutoConfiguration.ITEM_WRITE_EVENTS_LISTENER)) {
chunkProcessor.registerListener((ItemWriteListener) this.applicationContext
.getBean(BatchEventAutoConfiguration.ITEM_WRITE_EVENTS_LISTENER));
}
}
private void registerSkipEvents(SimpleChunkProvider chunkProvider) {
if (this.applicationContext.containsBean(BatchEventAutoConfiguration.SKIP_EVENTS_LISTENER)) {
chunkProvider.registerListener((SkipListener) this.applicationContext.getBean(BatchEventAutoConfiguration.SKIP_EVENTS_LISTENER));
if (this.applicationContext
.containsBean(BatchEventAutoConfiguration.SKIP_EVENTS_LISTENER)) {
chunkProvider.registerListener((SkipListener) this.applicationContext
.getBean(BatchEventAutoConfiguration.SKIP_EVENTS_LISTENER));
}
}
private void registerSkipEvents(SimpleChunkProcessor chunkProcessor) {
if(this.applicationContext.containsBean(BatchEventAutoConfiguration.SKIP_EVENTS_LISTENER)) {
chunkProcessor.registerListener((SkipListener) this.applicationContext.getBean(BatchEventAutoConfiguration.SKIP_EVENTS_LISTENER));
if (this.applicationContext
.containsBean(BatchEventAutoConfiguration.SKIP_EVENTS_LISTENER)) {
chunkProcessor.registerListener((SkipListener) this.applicationContext
.getBean(BatchEventAutoConfiguration.SKIP_EVENTS_LISTENER));
}
}
private void registerChunkEventsListener(Object bean) {
if(this.applicationContext.containsBean(BatchEventAutoConfiguration.CHUNK_EVENTS_LISTENER))
{
((TaskletStep)bean).registerChunkListener((ChunkListener)
this.applicationContext.getBean(BatchEventAutoConfiguration.CHUNK_EVENTS_LISTENER));
if (this.applicationContext
.containsBean(BatchEventAutoConfiguration.CHUNK_EVENTS_LISTENER)) {
((TaskletStep) bean)
.registerChunkListener((ChunkListener) this.applicationContext
.getBean(BatchEventAutoConfiguration.CHUNK_EVENTS_LISTENER));
}
}
private void registerJobExecutionEventListener(Object bean) {
if (bean instanceof AbstractJob &&
this.applicationContext.containsBean(BatchEventAutoConfiguration.JOB_EXECUTION_EVENTS_LISTENER)) {
JobExecutionListener jobExecutionEventsListener =
(JobExecutionListener) this.applicationContext.getBean(
BatchEventAutoConfiguration.JOB_EXECUTION_EVENTS_LISTENER);
if (bean instanceof AbstractJob && this.applicationContext.containsBean(
BatchEventAutoConfiguration.JOB_EXECUTION_EVENTS_LISTENER)) {
JobExecutionListener jobExecutionEventsListener = (JobExecutionListener) this.applicationContext
.getBean(BatchEventAutoConfiguration.JOB_EXECUTION_EVENTS_LISTENER);
AbstractJob job = (AbstractJob) bean;
job.registerJobExecutionListener(
jobExecutionEventsListener);
job.registerJobExecutionListener(jobExecutionEventsListener);
}
}
private void registerStepExecutionEventListener(Object bean) {
if (this.applicationContext.containsBean(BatchEventAutoConfiguration.STEP_EXECUTION_EVENTS_LISTENER)) {
StepExecutionListener stepExecutionListener =
(StepExecutionListener) this.applicationContext.getBean(BatchEventAutoConfiguration.STEP_EXECUTION_EVENTS_LISTENER);
if (this.applicationContext.containsBean(
BatchEventAutoConfiguration.STEP_EXECUTION_EVENTS_LISTENER)) {
StepExecutionListener stepExecutionListener = (StepExecutionListener) this.applicationContext
.getBean(BatchEventAutoConfiguration.STEP_EXECUTION_EVENTS_LISTENER);
AbstractStep step = (AbstractStep) bean;
step.registerStepExecutionListener(stepExecutionListener);
}
}
}

61
spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/support/TaskEventProperties.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2017 the original author or authors.
* Copyright 2015-2019 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.
@@ -20,48 +20,48 @@ import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.core.Ordered;
/**
* @author Ali Shahbour
* @author Ali Shahbour
*/
@ConfigurationProperties(prefix = "spring.cloud.task.batch.events")
public class TaskEventProperties {
/**
* Properties for jobExecution listener order
*/
/**
* Properties for jobExecution listener order.
*/
private int jobExecutionOrder = Ordered.LOWEST_PRECEDENCE;
/**
* Properties for stepExecution listener order
*/
/**
* Properties for stepExecution listener order.
*/
private int stepExecutionOrder = Ordered.LOWEST_PRECEDENCE;
/**
* Properties for itemRead listener order
*/
/**
* Properties for itemRead listener order.
*/
private int itemReadOrder = Ordered.LOWEST_PRECEDENCE;
/**
* Properties for itemProcess listener order
*/
/**
* Properties for itemProcess listener order.
*/
private int itemProcessOrder = Ordered.LOWEST_PRECEDENCE;
/**
* Properties for itemWrite listener order
*/
/**
* Properties for itemWrite listener order.
*/
private int itemWriteOrder = Ordered.LOWEST_PRECEDENCE;
/**
* Properties for chunk listener order
*/
/**
* Properties for chunk listener order.
*/
private int chunkOrder = Ordered.LOWEST_PRECEDENCE;
/**
* Properties for skip listener order
*/
/**
* Properties for skip listener order.
*/
private int skipOrder = Ordered.LOWEST_PRECEDENCE;
public int getJobExecutionOrder() {
return jobExecutionOrder;
return this.jobExecutionOrder;
}
public void setJobExecutionOrder(int jobExecutionOrder) {
@@ -69,7 +69,7 @@ public class TaskEventProperties {
}
public int getStepExecutionOrder() {
return stepExecutionOrder;
return this.stepExecutionOrder;
}
public void setStepExecutionOrder(int stepExecutionOrder) {
@@ -77,7 +77,7 @@ public class TaskEventProperties {
}
public int getItemReadOrder() {
return itemReadOrder;
return this.itemReadOrder;
}
public void setItemReadOrder(int itemReadOrder) {
@@ -85,7 +85,7 @@ public class TaskEventProperties {
}
public int getItemProcessOrder() {
return itemProcessOrder;
return this.itemProcessOrder;
}
public void setItemProcessOrder(int itemProcessOrder) {
@@ -93,7 +93,7 @@ public class TaskEventProperties {
}
public int getItemWriteOrder() {
return itemWriteOrder;
return this.itemWriteOrder;
}
public void setItemWriteOrder(int itemWriteOrder) {
@@ -101,7 +101,7 @@ public class TaskEventProperties {
}
public int getChunkOrder() {
return chunkOrder;
return this.chunkOrder;
}
public void setChunkOrder(int chunkOrder) {
@@ -109,10 +109,11 @@ public class TaskEventProperties {
}
public int getSkipOrder() {
return skipOrder;
return this.skipOrder;
}
public void setSkipOrder(int skipOrder) {
this.skipOrder = skipOrder;
}
}

89
spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/launcher/TaskLaunchRequest.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016 the original author or authors.
* Copyright 2015-2019 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.
@@ -16,7 +16,6 @@
package org.springframework.cloud.task.launcher;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.Collections;
@@ -34,13 +33,18 @@ import org.springframework.util.StringUtils;
*
* @author Glenn Renfro
*/
public class TaskLaunchRequest implements Serializable{
public class TaskLaunchRequest implements Serializable {
private static final long serialVersionUID = 1L;
private String uri;
private List<String> commandlineArguments;
private Map<String, String> environmentProperties;
private Map<String, String> deploymentProperties;
private String applicationName;
/**
@@ -49,20 +53,21 @@ public class TaskLaunchRequest implements Serializable{
* @param commandlineArguments list of commandlineArguments to be used by the task
* @param environmentProperties are the environment variables for this task.
* @param deploymentProperties are the variables used to setup task on the platform.
* @param applicationName name to be applied to the launched task. If set
* to null then the launched task name will be "Task-`hash code of the
* TaskLaunchRequest`.
* @param applicationName name to be applied to the launched task. If set to null then
* the launched task name will be "Task-`hash code of the TaskLaunchRequest`.
*/
public TaskLaunchRequest(String uri, List<String> commandlineArguments,
Map<String, String> environmentProperties,
Map<String, String> deploymentProperties,
String applicationName) {
Map<String, String> environmentProperties,
Map<String, String> deploymentProperties, String applicationName) {
Assert.hasText(uri, "uri must not be empty nor null.");
this.uri = uri;
this.commandlineArguments = (commandlineArguments == null) ? new ArrayList<String>() : commandlineArguments;
this.environmentProperties = environmentProperties == null ? new HashMap<String, String>() : environmentProperties;
this.deploymentProperties = deploymentProperties == null ? new HashMap<String, String>() : deploymentProperties;
this.commandlineArguments = (commandlineArguments == null) ? new ArrayList<>()
: commandlineArguments;
this.environmentProperties = environmentProperties == null ? new HashMap<>()
: environmentProperties;
this.deploymentProperties = deploymentProperties == null ? new HashMap<>()
: deploymentProperties;
setApplicationName(applicationName);
}
@@ -75,17 +80,17 @@ public class TaskLaunchRequest implements Serializable{
}
/**
* @return the current uri to the artifact for this launch request.
* @return the current uri to the artifact for this launch request.
*/
public String getUri() {
return uri;
return this.uri;
}
/**
* @return an unmodifiable list of arguments that will be used for the task execution
* @return an unmodifiable list of arguments that will be used for the task execution
*/
public List<String> getCommandlineArguments() {
return Collections.unmodifiableList(commandlineArguments);
return Collections.unmodifiableList(this.commandlineArguments);
}
/**
@@ -94,80 +99,76 @@ public class TaskLaunchRequest implements Serializable{
*/
public Map<String, String> getEnvironmentProperties() {
return environmentProperties;
return this.environmentProperties;
}
/**
* Returns the properties used by a {@link org.springframework.cloud.deployer.spi.task.TaskLauncher}
*
* Returns the properties used by a
* {@link org.springframework.cloud.deployer.spi.task.TaskLauncher}.
* @return deployment properties
*/
public Map<String, String> getDeploymentProperties() {
return deploymentProperties;
return this.deploymentProperties;
}
/**
* Returns the name that will be associated with the launched task.
*
* @return string containing the application name.
*/
public String getApplicationName() {
return applicationName;
return this.applicationName;
}
/**
* Sets the name to be applied to the launched task. If set
* to null then the launched task name will be "Task-`unique id`".
*
* Sets the name to be applied to the launched task. If set to null then the launched
* task name will be "Task-`unique id`".
* @param applicationName the name to be
*/
public void setApplicationName(String applicationName) {
this.applicationName = !StringUtils.hasText(applicationName) ? "Task-" +
UUID.randomUUID().toString() : applicationName;
this.applicationName = !StringUtils.hasText(applicationName)
? "Task-" + UUID.randomUUID().toString() : applicationName;
}
@Override
public String toString() {
return "TaskLaunchRequest{" +
"uri='" + uri + '\'' +
", commandlineArguments=" + commandlineArguments +
", environmentProperties=" + environmentProperties +
", deploymentProperties=" + deploymentProperties +
'}';
return "TaskLaunchRequest{" + "uri='" + this.uri + '\''
+ ", commandlineArguments=" + this.commandlineArguments
+ ", environmentProperties=" + this.environmentProperties
+ ", deploymentProperties=" + this.deploymentProperties + '}';
}
@Override
public boolean equals(Object o) {
if (this == o){
if (this == o) {
return true;
}
if (o == null || getClass() != o.getClass()){
if (o == null || getClass() != o.getClass()) {
return false;
}
TaskLaunchRequest that = (TaskLaunchRequest) o;
if (!uri.equals(that.uri)){
if (!this.uri.equals(that.uri)) {
return false;
}
if (!commandlineArguments.equals(that.commandlineArguments)){
if (!this.commandlineArguments.equals(that.commandlineArguments)) {
return false;
}
if(!deploymentProperties.equals(that.deploymentProperties))
{
if (!this.deploymentProperties.equals(that.deploymentProperties)) {
return false;
}
return environmentProperties.equals(that.environmentProperties);
return this.environmentProperties.equals(that.environmentProperties);
}
@Override
public int hashCode() {
final int HASH_DEFAULT = 31;
int result = uri.hashCode();
result = HASH_DEFAULT * result + commandlineArguments.hashCode();
result = HASH_DEFAULT * result + environmentProperties.hashCode();
result = HASH_DEFAULT * result + deploymentProperties.hashCode();
int result = this.uri.hashCode();
result = HASH_DEFAULT * result + this.commandlineArguments.hashCode();
result = HASH_DEFAULT * result + this.environmentProperties.hashCode();
result = HASH_DEFAULT * result + this.deploymentProperties.hashCode();
return result;
}
}

20
spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/launcher/TaskLauncherSink.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2018 the original author or authors.
* Copyright 2015-2019 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.
@@ -30,7 +30,6 @@ import org.springframework.core.io.Resource;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.util.Assert;
/**
* A sink stream application that launches a tasks.
*
@@ -42,20 +41,23 @@ public class TaskLauncherSink {
private final static Logger logger = LoggerFactory.getLogger(TaskLauncherSink.class);
// @checkstyle:off
@Autowired
public TaskLauncher taskLauncher;
// @checkstyle:on
@Autowired
private DelegatingResourceLoader resourceLoader;
/**
* Launches a task upon the receipt of a valid TaskLaunchRequest.
* @param taskLaunchRequest is a TaskLaunchRequest containing the information required to launch
* a task.
* @param taskLaunchRequest is a TaskLaunchRequest containing the information required
* to launch a task.
* @throws Exception if error occurs during task launch.
*/
@ServiceActivator(inputChannel = Sink.INPUT)
public void taskLauncherSink(TaskLaunchRequest taskLaunchRequest) throws Exception{
public void taskLauncherSink(TaskLaunchRequest taskLaunchRequest) throws Exception {
launchTask(taskLaunchRequest);
}
@@ -63,8 +65,12 @@ public class TaskLauncherSink {
Assert.notNull(this.taskLauncher, "TaskLauncher has not been initialized");
logger.info("Launching Task for the following uri " + taskLaunchRequest.getUri());
Resource resource = this.resourceLoader.getResource(taskLaunchRequest.getUri());
AppDefinition definition = new AppDefinition(taskLaunchRequest.getApplicationName(), taskLaunchRequest.getEnvironmentProperties());
AppDeploymentRequest request = new AppDeploymentRequest(definition, resource, taskLaunchRequest.getDeploymentProperties(), taskLaunchRequest.getCommandlineArguments());
AppDefinition definition = new AppDefinition(
taskLaunchRequest.getApplicationName(),
taskLaunchRequest.getEnvironmentProperties());
AppDeploymentRequest request = new AppDeploymentRequest(definition, resource,
taskLaunchRequest.getDeploymentProperties(),
taskLaunchRequest.getCommandlineArguments());
this.taskLauncher.launch(request);
}

11
spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/launcher/annotation/EnableTaskLauncher.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016 the original author or authors.
* Copyright 2015-2019 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.
@@ -42,12 +42,12 @@ import org.springframework.context.annotation.Import;
* &#064;Bean
* public MyCommandLineRunner myCommandLineRunner() {
* return new MyCommandLineRunner()
* }
* }
* }
* </pre>
*
* Note that only one of your configuration classes needs to have the <code>&#064;EnableTaskLauncher</code>
* annotation.
* Note that only one of your configuration classes needs to have the
* <code>&#064;EnableTaskLauncher</code> annotation.
*
* @author Glenn Renfro
*/
@@ -55,6 +55,7 @@ import org.springframework.context.annotation.Import;
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Inherited
@Import({TaskLauncherSink.class})
@Import({ TaskLauncherSink.class })
public @interface EnableTaskLauncher {
}

36
spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/listener/TaskEventAutoConfiguration.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2018 the original author or authors.
* Copyright 2015-2019 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.
@@ -13,6 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.task.listener;
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
@@ -31,38 +32,51 @@ import org.springframework.integration.gateway.GatewayProxyFactoryBean;
import org.springframework.messaging.MessageChannel;
/**
*
* @author Michael Minella
*/
@Configuration
@ConditionalOnClass(EnableBinding.class)
@ConditionalOnBean(TaskLifecycleListener.class)
// @checkstyle:off
@ConditionalOnProperty(prefix = "spring.cloud.task.events", name = "enabled", havingValue = "true", matchIfMissing = true)
// @checkstyle:on
@PropertySource("classpath:/org/springframework/cloud/task/application.properties")
@AutoConfigureBefore(BindingServiceConfiguration.class)
@AutoConfigureAfter(SimpleTaskAutoConfiguration.class)
public class TaskEventAutoConfiguration {
/**
* Task Event channels definition.
*/
public interface TaskEventChannels {
/**
* Name of the task events channel.
*/
String TASK_EVENTS = "task-events";
@Output(TASK_EVENTS)
MessageChannel taskEvents();
}
/**
* Configuration for a {@link TaskExecutionListener}.
*/
@Configuration
@EnableBinding(TaskEventChannels.class)
public static class ListenerConfiguration {
@Bean
public GatewayProxyFactoryBean taskEventListener() {
GatewayProxyFactoryBean factoryBean =
new GatewayProxyFactoryBean(TaskExecutionListener.class);
GatewayProxyFactoryBean factoryBean = new GatewayProxyFactoryBean(
TaskExecutionListener.class);
factoryBean.setDefaultRequestChannelName(TaskEventChannels.TASK_EVENTS);
return factoryBean;
}
}
public interface TaskEventChannels {
String TASK_EVENTS = "task-events";
@Output(TASK_EVENTS)
MessageChannel taskEvents();
}
}

16
spring-cloud-task-stream/src/main/resources/META-INF/additional-spring-configuration-metadata.json
View File

@@ -1,10 +1,10 @@
{
"properties": [
{
"defaultValue": true,
"name": "spring.cloud.task.events.enabled",
"description": "This property is used to determine if a task app should emit task events.",
"type": "java.lang.Boolean"
}
]
"properties": [
{
"defaultValue": true,
"name": "spring.cloud.task.events.enabled",
"description": "This property is used to determine if a task app should emit task events.",
"type": "java.lang.Boolean"
}
]
}