Build a feed stream system for tens of millions of users

Step 1: Initialize the Timeline model

Use the Timeline SDK to initialize the store (outbox), sync (inbox), and meta store separately. Configure a search index for the store to support queries by sender and message content.

TimelineStoreFactory serviceFactory = new TimelineStoreFactoryImpl(syncClient);

// Initialize store (outbox): stores all published messages
TimelineIdentifierSchema idSchema = new TimelineIdentifierSchema.Builder()
        .addStringField("timeline_id").build();

IndexSchema timelineIndex = new IndexSchema();
timelineIndex.addFieldSchema(new FieldSchema("sender", FieldType.KEYWORD)
        .setIndex(true).setEnableSortAndAgg(true).setStore(true));
timelineIndex.addFieldSchema(new FieldSchema("send_time", FieldType.LONG)
        .setIndex(true).setEnableSortAndAgg(true).setStore(true));
timelineIndex.addFieldSchema(new FieldSchema("text", FieldType.TEXT)
        .setIndex(true).setStore(true).setAnalyzer(FieldSchema.Analyzer.MaxWord));

TimelineSchema storeSchema = new TimelineSchema("feed_timeline_store_table", idSchema)
        .autoGenerateSeqId()
        .setTimeToLive(-1)
        .withIndex("feed_timeline_store_index", timelineIndex);

TimelineStore store = serviceFactory.createTimelineStore(storeSchema);

// Initialize sync (inbox): receives pushed messages from followed users
TimelineSchema syncSchema = new TimelineSchema("feed_timeline_sync_table", idSchema)
        .autoGenerateSeqId()
        .setTimeToLive(-1);

TimelineStore sync = serviceFactory.createTimelineStore(syncSchema);

// Initialize meta: stores timeline metadata
IndexSchema metaIndex = new IndexSchema();
metaIndex.setFieldSchemas(Arrays.asList(
        new FieldSchema("type", FieldType.KEYWORD)
                .setIndex(true).setEnableSortAndAgg(true).setStore(true),
        new FieldSchema("user", FieldType.TEXT)
                .setIndex(true).setStore(true).setAnalyzer(FieldSchema.Analyzer.MaxWord)
));

TimelineMetaSchema metaSchema = new TimelineMetaSchema("feed_timeline_meta_table", idSchema)
        .withIndex("feed_timeline_meta_index", metaIndex);

TimelineMetaStore metaStore = serviceFactory.createMetaStore(metaSchema);

// Create all tables (call once during initialization)
store.prepareTables();
sync.prepareTables();
metaStore.prepareTables();

Step 2: Follow a user

Call the PutRow operation of the base SDK to write a record to the user relationship table and establish a follow relationship.

public void follow(String userA, String userB) {
    PrimaryKey primaryKey = PrimaryKeyBuilder.createPrimaryKeyBuilder()
            .addPrimaryKeyColumn("main_user", PrimaryKeyValue.fromString(userA))
            .addPrimaryKeyColumn("sub_user", PrimaryKeyValue.fromString(userB))
            .build();

    RowPutChange rowPutChange = new RowPutChange("feed_user_relation_table", primaryKey);
    rowPutChange.addColumn("timestamp", ColumnValue.fromLong(System.currentTimeMillis()));

    syncClient.putRow(new PutRowRequest(rowPutChange));
}

Step 3: Publish a message

To publish a message, write it to the sender's store (outbox) first, then push it to all followers' inboxes (sync). This is the push operation in the combined push-pull model.

public void sendMessage(String userId, Message message, List<String> followers) {
    // 1. Write to sender's store (outbox)
    TimelineIdentifier identifier = new TimelineIdentifier.Builder()
            .addField("timeline_id", message.getTimelineId())
            .build();
    store.createTimelineQueue(identifier).store(message.getTimelineMessage());

    // 2. Push to each follower's sync (inbox)
    for (String friend : followers) {
        TimelineIdentifier friendId = new TimelineIdentifier.Builder()
                .addField("timeline_id", friend)
                .build();
        sync.createTimelineQueue(friendId).store(message.getTimelineMessage());
    }
}

Step 4: Read the feed stream

To read the feed stream, scan messages from the sync library (inbox) sequentially by auto-increment sequence ID. This is the pull operation in the combined push-pull model. The client records the lastSequenceId from the previous read and resumes scanning from that position in the next request.

public List<Post> fetchFeed(String userId, long lastSequenceId) {
    TimelineIdentifier identifier = new TimelineIdentifier.Builder()
            .addField("timeline_id", userId)
            .build();

    ScanParameter parameter = new ScanParameter()
            .scanForward(lastSequenceId)
            .maxCount(30);

    Iterator<TimelineEntry> iterator = sync.createTimelineQueue(identifier).scan(parameter);

    List<Post> posts = new LinkedList<>();
    int counter = 0;
    while (iterator.hasNext() && counter++ <= 30) {
        TimelineEntry entry = iterator.next();
        TimelineMessage msg = entry.getMessage();

        Post post = new Post()
                .setName(msg.getString("sender"))
                .setTime(msg.getLong("send_time"))
                .setContent(msg.getString("text"))
                .setSequenceId(entry.getSequenceID());
        posts.add(post);
    }
    return posts;
}

To view all messages published by a specific user (user profile page), scan from the store instead of the sync library. The code structure is the same.