Skip to content

Commit 8423609

Browse files
acktsapacktsap
authored
Update doc for reader-processor
Signed-off-by: Taeik Lim <[email protected]>
1 parent 0408b7e commit 8423609

8 files changed

+1219
-16
lines changed

‎doc/en/step/item-stream-flux-reader-processor-writer.md‎

Lines changed: 152 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -6,9 +6,12 @@
66
- [Create a tasklet without a processor](#create-a-tasklet-without-a-processor)
77
- [Java](#java-1)
88
- [Kotlin](#kotlin-1)
9-
- [Use a callback](#use-a-callback)
9+
- [Create a tasklet without a writer](#create-a-tasklet-without-a-writer)
1010
- [Java](#java-2)
1111
- [Kotlin](#kotlin-2)
12+
- [Use a callback](#use-a-callback)
13+
- [Java](#java-3)
14+
- [Kotlin](#kotlin-3)
1215

1316
Spring uses a reactive library called [Reactor](https://projectreactor.io/), which provides streams for various types of data using `Flux`. For `ItemReader` in Spring Batch to use the data read using `Flux`, you need to extract each item from `Flux` and return it.
1417

@@ -326,9 +329,156 @@ open class TestJobConfig(
326329
}
327330
```
328331

332+
## Create a tasklet without a writer
333+
334+
If you need only `ItemStreamReader` and `ItemProcessor` without a writer, you can inherit `ItemStreamFluxReaderProcessor` to define `ItemStreamReader` and `ItemProcessor` in a single class.
335+
336+
### Java
337+
338+
In Java, you can convert a tasklet defined using `AdapterFactory` to `ItemStreamReader` and `ItemProcessor`.
339+
340+
```java
341+
@Component
342+
@StepScope
343+
class SampleTasklet implements ItemStreamFluxReaderProcessor<Integer, String> {
344+
345+
@Value("#{jobParameters['totalCount']}")
346+
private long totalCount;
347+
348+
private int count = 0;
349+
350+
@NonNull
351+
@Override
352+
public Flux<? extends Integer> readFlux(@NonNull ExecutionContext executionContext) {
353+
System.out.println("totalCount: " + totalCount);
354+
return Flux.generate(sink -> {
355+
if (count < totalCount) {
356+
sink.next(count);
357+
++count;
358+
} else {
359+
sink.complete();
360+
}
361+
});
362+
}
363+
364+
@Override
365+
public String process(@NonNull Integer item) {
366+
return "'" + item.toString() + "'";
367+
}
368+
}
369+
```
370+
371+
```java
372+
@Configuration
373+
public class TestJobConfig {
374+
375+
@Bean
376+
public Job testJob(
377+
SampleTasklet sampleTasklet,
378+
JobRepository jobRepository,
379+
PlatformTransactionManager transactionManager
380+
) {
381+
return new JobBuilder("testJob", jobRepository)
382+
.start(
383+
new StepBuilder("testStep", jobRepository)
384+
.<Integer, String>chunk(3, transactionManager)
385+
.reader(AdapterFactory.itemStreamReader(sampleTasklet))
386+
.processor(AdapterFactory.itemProcessor(sampleTasklet))
387+
.writer(chunk -> System.out.println(chunk.getItems()))
388+
.build()
389+
)
390+
.build();
391+
}
392+
}
393+
```
394+
395+
You can statically import the method of `AdapterFactory` for better readability.
396+
397+
```java
398+
import static com.navercorp.spring.batch.plus.step.AdapterFactory.itemStreamReader;
399+
import static com.navercorp.spring.batch.plus.step.AdapterFactory.itemStreamWriter;
400+
401+
...
402+
403+
@Configuration
404+
public class TestJobConfig {
405+
406+
@Bean
407+
public Job testJob(
408+
SampleTasklet sampleTasklet,
409+
JobRepository jobRepository,
410+
PlatformTransactionManager transactionManager
411+
) {
412+
return new JobBuilder("testJob", jobRepository)
413+
.start(
414+
new StepBuilder("testStep", jobRepository)
415+
.<Integer, String>chunk(3, transactionManager)
416+
.reader(itemStreamReader(sampleTasklet))
417+
.processor(itemProcessor(sampleTasklet))
418+
.writer(chunk -> System.out.println(chunk.getItems()))
419+
.build()
420+
)
421+
.build();
422+
}
423+
}
424+
```
425+
426+
### Kotlin
427+
428+
In Kotlin, you can easily convert a tasklet defined using an extension function in Spring Batch Plus to `ItemStreamReader` and `ItemProcessor`.
429+
430+
```Kotlin
431+
@Component
432+
@StepScope
433+
open class SampleTasklet(
434+
@Value("#{jobParameters['totalCount']}") private var totalCount: Long,
435+
) : ItemStreamFluxReaderProcessor<Int, String> {
436+
private var count = 0
437+
438+
override fun readFlux(executionContext: ExecutionContext): Flux<out Int> {
439+
println("totalCount: $totalCount")
440+
return Flux.generate { sink ->
441+
if (count < totalCount) {
442+
sink.next(count)
443+
++count
444+
} else {
445+
sink.complete()
446+
}
447+
}
448+
}
449+
450+
override fun process(item: Int): String? {
451+
return "'$item'"
452+
}
453+
}
454+
```
455+
456+
```Kotlin
457+
@Configuration
458+
open class TestJobConfig(
459+
private val batch: BatchDsl,
460+
private val transactionManager: PlatformTransactionManager,
461+
) {
462+
@Bean
463+
open fun testJob(
464+
sampleTasklet: SampleTasklet,
465+
): Job = batch {
466+
job("testJob") {
467+
step("testStep") {
468+
chunk<Int, String>(3, transactionManager) {
469+
reader(sampleTasklet.asItemStreamReader())
470+
processor(sampleTasklet.asItemProcessor())
471+
writer { chunk -> println(chunk.items) }
472+
}
473+
}
474+
}
475+
}
476+
}
477+
```
478+
329479
## Use a callback
330480

331-
You can define a callback method for `ItemStream` of `ItemStreamWriter` in `ItemStreamFluxReaderProcessorWriter` and `ItemStreamFluxReaderWriter`. You can selectively define a callback method.
481+
You can define a callback method for `ItemStream`. You can selectively define a callback method.
332482

333483
### Java
334484

‎doc/en/step/item-stream-iterable-reader-processor-writer.md‎

Lines changed: 158 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -6,9 +6,12 @@
66
- [Create a tasklet without a processor](#create-a-tasklet-without-a-processor)
77
- [Java](#java-1)
88
- [Kotlin](#kotlin-1)
9-
- [Use a callback](#use-a-callback)
9+
- [Create a tasklet without a writer](#create-a-tasklet-without-a-writer)
1010
- [Java](#java-2)
1111
- [Kotlin](#kotlin-2)
12+
- [Use a callback](#use-a-callback)
13+
- [Java](#java-3)
14+
- [Kotlin](#kotlin-3)
1215

1316
A chunk-oriented step in Spring Batch consists of `ItemReader`, `ItemProcessor`, and `ItemWriter`, which are usually defined separately and then assembled to define a step. However, there are some issues with this approach: it is difficult to share data between `ItemReader`, `ItemProcessor`, and `ItemWriter`, and you need to see each respective file to understand the batch flow. Also, if the classes are not reused, they can make the elements of a job less coherent.
1417

@@ -328,9 +331,162 @@ open class TestJobConfig(
328331
}
329332
```
330333

334+
## Create a tasklet without a writer
335+
336+
If you need only `ItemStreamReader` and `ItemProcessor` without a writer, you can inherit `ItemStreamIterableReaderProcessor` to define `ItemStreamReader` and `ItemProcessor` in a single class.
337+
338+
### Java
339+
340+
In Java, you can convert a tasklet defined using `AdapterFactory` to `ItemStreamReader` and `ItemProcessor`.
341+
342+
```java
343+
@Component
344+
@StepScope
345+
class SampleTasklet implements ItemStreamIterableReaderProcessor<Integer, String> {
346+
347+
@Value("#{jobParameters['totalCount']}")
348+
private long totalCount;
349+
350+
private int count = 0;
351+
352+
@NonNull
353+
@Override
354+
public Iterable<? extends Integer> readIterable(@NonNull ExecutionContext executionContext) {
355+
System.out.println("totalCount: " + totalCount);
356+
return () -> new Iterator<>() {
357+
@Override
358+
public boolean hasNext() {
359+
return count < totalCount;
360+
}
361+
362+
@Override
363+
public Integer next() {
364+
return count++;
365+
}
366+
};
367+
}
368+
369+
@Override
370+
public String process(@NonNull Integer item) {
371+
return "'" + item.toString() + "'";
372+
}
373+
}
374+
```
375+
376+
```java
377+
@Configuration
378+
public class TestJobConfig {
379+
380+
@Bean
381+
public Job testJob(
382+
SampleTasklet sampleTasklet,
383+
JobRepository jobRepository,
384+
PlatformTransactionManager transactionManager
385+
) {
386+
return new JobBuilder("testJob", jobRepository)
387+
.start(
388+
new StepBuilder("testStep", jobRepository)
389+
.<Integer, String>chunk(3, transactionManager)
390+
.reader(AdapterFactory.itemStreamReader(sampleTasklet))
391+
.processor(AdapterFactory.itemProcessor(sampleTasklet))
392+
.writer(chunk -> System.out.println(chunk.getItems()))
393+
.build()
394+
)
395+
.build();
396+
}
397+
}
398+
```
399+
400+
You can statically import the method of `AdapterFactory` for better readability.
401+
402+
```java
403+
import static com.navercorp.spring.batch.plus.step.AdapterFactory.itemStreamReader;
404+
import static com.navercorp.spring.batch.plus.step.AdapterFactory.itemStreamWriter;
405+
406+
...
407+
408+
@Configuration
409+
public class TestJobConfig {
410+
411+
@Bean
412+
public Job testJob(
413+
SampleTasklet sampleTasklet,
414+
JobRepository jobRepository,
415+
PlatformTransactionManager transactionManager
416+
) {
417+
return new JobBuilder("testJob", jobRepository)
418+
.start(
419+
new StepBuilder("testStep", jobRepository)
420+
.<Integer, String>chunk(3, transactionManager)
421+
.reader(itemStreamReader(sampleTasklet))
422+
.processor(itemProcessor(sampleTasklet))
423+
.writer(chunk -> System.out.println(chunk.getItems()))
424+
.build()
425+
)
426+
.build();
427+
}
428+
}
429+
```
430+
431+
### Kotlin
432+
433+
In Kotlin, you can easily convert a tasklet defined using an extension function in Spring Batch Plus to `ItemStreamReader` and `ItemProcessor`.
434+
435+
```Kotlin
436+
@Component
437+
@StepScope
438+
open class SampleTasklet(
439+
@Value("#{jobParameters['totalCount']}") private var totalCount: Long,
440+
) : ItemStreamIterableReaderProcessor<Int, String> {
441+
private var count = 0
442+
443+
override fun readIterable(executionContext: ExecutionContext): Iterable<Int> {
444+
println("totalCount: $totalCount")
445+
return Iterable {
446+
object : Iterator<Int> {
447+
override fun hasNext(): Boolean {
448+
return count < totalCount
449+
}
450+
451+
override fun next(): Int {
452+
return count++
453+
}
454+
}
455+
}
456+
}
457+
458+
override fun process(item: Int): String? {
459+
return "'$item'"
460+
}
461+
}
462+
```
463+
464+
```Kotlin
465+
@Configuration
466+
open class TestJobConfig(
467+
private val batch: BatchDsl,
468+
private val transactionManager: PlatformTransactionManager,
469+
) {
470+
@Bean
471+
open fun testJob(
472+
sampleTasklet: SampleTasklet,
473+
): Job = batch {
474+
job("testJob") {
475+
step("testStep") {
476+
chunk<Int, String>(3, transactionManager) {
477+
reader(sampleTasklet.asItemStreamReader())
478+
processor(sampleTasklet.asItemProcessor())
479+
writer { chunk -> println(chunk.items) }
480+
}
481+
}
482+
}
483+
}
484+
}
485+
```
486+
331487
## Use a callback
332488

333-
You can define a callback method for `ItemStream` of `ItemStreamWriter` in `ItemStreamIterableReaderProcessorWriter` and `ItemStreamIterableReaderWriter`. You can selectively define a callback method.
489+
You can define a callback method for `ItemStream`. You can selectively define a callback method.
334490

335491
### Java
336492

0 commit comments

Comments
 (0)