Custom Connectors

If Hazelcast doesn't natively support the data source/sink you need, you can build a connector for it yourself by using the SourceBuilder and SinkBuilder.

SourceBuilder

To make a custom source connector you need two basic ingredients:

• a context object that holds all the resources and state you need to keep track of
• a stateless function, `fillBufferFn`, taking two parameters: the state object and a buffer object provided by the Jet API

Hazelcast repeatedly calls fillBufferFn whenever it needs more data items. Optimally, the function will fill the buffer with the items it can acquire without blocking. A hundred items at a time is enough to eliminate any per-call overheads within Hazelcast. The function is allowed to block as well, but taking longer than a second to complete can have negative effects on the overall performance of the processing pipeline.

In the following examples we build a simple batch source that emits the lines of a file:

java

BatchSource<String> fileSource = SourceBuilder
    .batch("file-source", x -> new BufferedReader(new FileReader("input.txt")))
    .<String>fillBufferFn((in, buf) -> {
        String line = in.readLine();
        if (line != null) {
            buf.add(line);
        } else {
            buf.close();
        }
    })
    .destroyFn(BufferedReader::close)
    .build();

For a more involved example (which reads data in batches for efficiency, deals with unbounded data, emits timestamps, is distributed and fault tolerant see the Custom Batch Sources and Custom Stream Sources tutorials).

SinkBuilder

To make your custom sink connector you need two basic ingredients:

• a context object that holds all the resources and state you need to keep track of
• a stateless function, `receiveFn`, taking two parameters: the state object and a data item sent to the sink

In the following example we build a simple sink which writes the toString() form of objects to a file:

java

Sink<Object> sink = sinkBuilder(
    "file-sink", x -> new PrintWriter(new FileWriter("output.txt")))
    .receiveFn((out, item) -> out.println(item.toString()))
    .destroyFn(PrintWriter::close)
    .build();

For a more involved example, covering issues like batching, distributiveness and fault tolerance, see the Custom Sinks tutorial).