< Prev - Examples - Next >

PL/SQL Batch/Stream Processor

Example Description

Important Notes !

PL/SQL Package Specification : BATCH_STREAM_PROCESSOR
create or replace package batch_stream_processor
/**
 * PL/SQL Batch/Stream processor demonstration.
 */
as

-- PL/SQL record row declaration for batch/stream processing
type rec_batch_row is record (
  no            number(9),
  description   varchar2(100),
  ts            timestamp
);

-- cursor definition
type c_batch_row is ref cursor return rec_batch_row;

/**
 * Functions returns one cursor for batch processing.
 *
 * @param i_number_of_rows Number of rows to generate.
 * @return Row of PL/SQL record
 */
function job1 (i_number_of_rows in number) return c_batch_row;

end batch_stream_processor;
PL/SQL Package Body : BATCH_STREAM_PROCESSOR
create or replace package body batch_stream_processor
as

function job1 (i_number_of_rows in number) return c_batch_row
is
  c c_batch_row;
begin
  -- generate dummy list of rows
  open c for
    select  level, 
            'any description no. ' || level, 
            systimestamp+level+level/12. 
      from  dual 
      connect by level<=i_number_of_rows;

  -- return cursor
  return c;
end job1;

end batch_stream_processor;
Java Calling Class : BatchStreamProcessor.java
package batch_stream_processor_examples;

import java.util.stream.Stream;

import factory.ExamplesRPCFactory;
import service.BatchStreamProcessorService;
import transferobject.BatchStreamProcessorTO;

public class BatchStreamProcessor {
  public static void main(String[] args) {
    try {
      // getting the service
      BatchStreamProcessorService service = ExamplesRPCFactory.getBatchStreamProcessorService();

      // use try-with-resources statement to free resources after call
      try (Stream<BatchStreamProcessorTO.RecBatchRow> stream = service.job1Stream(1000, 0)) {
        stream.forEach(e -> System.out.format("no:%d   description:%s   timestamp:%3$tD %3$tT%n", e.no, e.description, e.ts));
      }
    }
    catch (Exception e) {
      e.printStackTrace();
    }
  }
}