package com.emc.example.client;

import java.util.Map;

import com.emc.example.client.dummy.RichSinkFunction;
import com.emc.nautilus.streaming.Producer;
import com.emc.nautilus.streaming.Transaction;
import com.emc.nautilus.streaming.TxFailedException;

public class TransactionalStreamProducerSink<IN> extends RichSinkFunction<IN> // ...
{
	//
	private Transaction<IN> openTxn;
	Producer<IN> producer;

	public void invoke(IN value) throws TxFailedException {
		if (openTxn == null) {
			openTxn = producer.startTransaction( 60000);
			
		}
		String routingKey = getRoutingKey(value);
		openTxn.publish(routingKey, value);
	}

	// ...
	public void notifyCheckpointComplete() throws TxFailedException {
			openTxn.commit();
	}

	public byte[] snapshotState() {
		return serializeOpenTxns();
	}

	public void restoreState(byte[] state) {
		restoreOpenTxns(state);
	}

	// ...
	private String getRoutingKey(IN value) {
		// TODO Auto-generated method stub
		return null;
	}

	private byte[] serializeOpenTxns() {
		// TODO Auto-generated method stub
		return null;
	}

	private void restoreOpenTxns(byte[] state) {
		// TODO Auto-generated method stub

	}
}
@@ -523,17 +523,16 @@
 ion(
- 
 60000);%0A
 %09%09%09%0A
@@ -527,16 +527,16 @@
 60000);%0A
+
 %09%09%09%0A%09%09%7D%0A
@@ -779,17 +779,16 @@
 eOpenTxn
-s
 ();%0A%09%7D%0A%0A
@@ -845,17 +845,16 @@
 eOpenTxn
-s
 (state);
@@ -996,17 +996,16 @@
 eOpenTxn
-s
 () %7B%0A%09%09/
@@ -1049,32 +1049,32 @@
 eturn null;%0A%09%7D%0A%0A
+
 %09private void re
@@ -1085,17 +1085,16 @@
 eOpenTxn
-s
 (byte%5B%5D 

Update example.