Redis Streams placeholder implementation

redis-streams-overview.java

// Publish API for redis stream
void send(Map<String, String> messagePayload) {
  StreamMessageId messageId = redisClient.getStream(streamName, new StringCodec("UTF-8")).addAll(messagePayload, maxTopicSize);
  log.info("Message {} was inserted to stream", messageId);
}


// Read API for polling redis stream
List<Message> read(long maxWaitTime) {
  // If you are using Redis 6.2 or above, consider using XAUTOCLAIM instead of a combination of XPENDING and XCLAIM
  List<Message> pendingMessages = getPendingEntries();  
  if (messages.isEmpty()) {
    // No message in the pending queue
    return getNewMessages();
  } else {
    List<Message> claimedMessages = claimEntries(pendingMessages);
    if (!claimedMessages.isEmpty()) {
      return claimedMessages;
    } else {
      return getNewMessages(maxWaitTime);
    }
  }
}

// Consumer runnable thread
while (true) {
  messages = client.read(blockTimeout);
  for (Message message : messages) {
    // Do some custom processing based on your application reads
    processMessage(message);
    // Confirm that the message is received so that no other consumer re-reads it
    acknowledgeMessage(message);
  }
}