DataHub is a publish-subscribe(pub-sub) system developed for stream data processing.
DataHub enables you to build applications based on stream data analysis.
DataHub can be also used as the data entrance of StreamCompute and MaxCompute.

git clone ...
cd datahub-sdk-java
mvn clean install -DskipTests

<dependency>
    <groupId>com.aliyun.datahub</groupId>
    <artifactId>aliyun-sdk-datahub</artifactId>
    <version>2.1.1</version>
</dependency>

http://dh-cn-hangzhou.aliyuncs.com
http://dh-cn-hangzhou-internal.aliyuncs.com

DataHub is now on public testing, please contact us for creating projects.

Account account = new AliyunAccount("your access id", "your access key");
DatahubConfiguration conf = new DatahubConfiguration(account, "datahub endpoint");
DatahubClient client = new DatahubClient(conf);

int shardCount = 3;
int lifecycle = 7;
RecordType type = RecordType.TUPLE;
RecordSchema schema = new RecordSchema();
schema.addField(new Field("field1", FieldType.BIGINT));
schema.addField(new Field("field2", FieldType.String));
String comment = "The first topic";
client.createTopic(projectName, topicName, shardCount, lifecycle, type, schema, comment);

// construct the records to be upload
RecordSchema schema = client.getTopic("projectName", "topicName").getRecordSchema();
List<RecordEntry> recordEntries = new ArrayList<~>();
RecordEntry entry = new RecordEntry(schema);
entry.setBigint(0, 123456);
entry.setString(1, "string");
entry.setShardId("shardId");

entry.putAttribute("attr_id", "2902381718"); //Attribute is the optional value for the record
recordEntries.add(entry);

// put records
client.putRecords("projectName", "topicName", recordEntries);

// get records
GetCursorResult cursor = client.getCursor("projectName", "topicName", "shardId", 1455869335000 /*ms*/);
GetRecordsResult r = client.getRecords("projectName", "topicName", "shardId", cursor.getCursor(), 10);