create readme and simple basic run through

Author: jphaugla

README.md

@@ -1,2 +1,85 @@
 # redisSMSLuaA-A
-redis SMS with LUA active/active
+Provides a quick-start example of using Redis Streams for parsing incoming messages in an active/active redis enterprise.
+## Overview
+A java producer will produce messages on a redis stream across two redis clusters in an active/active database.
+A java consumer will consume from the stream and create redis structures from the streams messages.
+First, on simple consumer will be used but a lua consumer will be developed next to eliminate round trips to the client java application
+
+## Redis Advantages for message partition streaming
+ * Redis easily handles high write transaction volume
+ * Redis enterprise scales vertically (large nodes)  and horizontally (many nodes)
+ * Redis enterprise active/active allows adding to a stream from either "region"
+ * Redis allows putting TTL on the hash
+
+## Requirements
+* Docker installed on your local system, see [Docker Installation Instructions](https://docs.docker.com/engine/installation/).
+
+* When using Docker for Mac or Docker for Windows, the default resources allocated to the linux VM running docker are 2GB RAM and 2 CPU's. Make sure to adjust these resources to meet the resource requirements for the containers you will be running. More information can be found here on adjusting the resources allocated to docker.
+
+[Docker for mac](https://docs.docker.com/docker-for-mac/#advanced)
+
+[Docker for windows](https://docs.docker.com/docker-for-windows/#advanced)
+
+## Links that help!
+ * [Active/Active docker under crdt-application](https://github.com/RedisLabs/redis-for-dummies)
+ * [Getting Started with Redis Streams and java github](https://github.com/tgrall/redis-streams-101-java)
+ * [Redis Active/Active CLI reference](https://docs.redis.com/latest/rs/references/crdb-cli-reference/)
+## Create Redis Enterprise Active/Active database
+ * Prepare Docker environment-see the Prerequisites section above...
+ * Clone the github 
+```bash
+git clone https://github.com/jphaugla/redisSMSLuaA-A
+```
+ * Refer to the notes for redis Docker images used but don't get too bogged down as docker compose handles everything except for a few admin steps on tomcat.
+   * [https://hub.docker.com/r/redislabs/redis/](https://hub.docker.com/r/redislabs/redis/)
+ * Open terminal and change to the github home to create the clusters and the clustered database
+```bash
+./create_redis_enterprise_clusters.sh 
+./setup_redis_enterprise_clusters.sh
+./create_crdb.sh
+```
+To access the databases using redis-cli, leverage the two different port numbers
+```bash
+# this is first redis cluster
+redis-cli -p 12000
+# this is second redis cluster
+redis-cli -p 12002
+```
+## The java code
+## To execute the java code
+(Alternatively, this can be run through intellij)
+
+ * Compile the code
+```bash
+mvn clean verify
+```
+ *  run the consumer   
+```bash
+./runconsumer.sh
+```
+ *   run the producer
+```bash
+./runproducer.sh
+```
+## Verify the results
+```bash
+redis-cli -p 12000
+> keys *
+```
+### Should see something similar to this![redis keys](images/keys.png)
+* There is a Redis set for each message containing the hash key for each message part belonging to this message
+```bash
+type weather_sensor:wind:message:MSG0
+```
+returns "set"
+```bash
+smembers  weather_sensor:wind:message:MSG0
+```
+returns all the message parts for this message
+```bash
+hgetall weather_sensor:wind:hash:1640798739595-2
+```
+return all the hash key value pairs from the message part
+## Additional tooling
+scripts are provided to support typical [crdb-cli](https://docs.redis.com/latest/rs/references/crdb-cli-reference/) commands
+The crdpurge.sh is very handy to efficiently do a "purgeall" of the database contents

crdpurge.sh

@@ -0,0 +1,2 @@
+guid=$1
+docker exec -it rp1 bash -c "/opt/redislabs/bin/crdb-cli crdb flush --crdb-guid $guid"

images/keys.png

@@ -12,7 +12,7 @@
         <dependency>
             <groupId>io.lettuce</groupId>
             <artifactId>lettuce-core</artifactId>
-            <version>5.1.8.RELEASE</version>
+            <version>6.1.5.RELEASE</version>
         </dependency>
         <dependency>
             <groupId>org.junit.jupiter</groupId>
@@ -26,10 +26,10 @@
         <plugins>
             <plugin>
                 <artifactId>maven-compiler-plugin</artifactId>
-                <version>3.2</version>
+                <version>3.8.1</version>
                 <configuration>
-                    <source>1.8</source>
-                    <target>1.8</target>
+                    <source>8</source>
+                    <target>8</target>
                 </configuration>
             </plugin>
         </plugins>

pom.xml

@@ -4,13 +4,14 @@
 import io.lettuce.core.api.StatefulRedisConnection;
 import io.lettuce.core.api.sync.RedisCommands;
 
-
-import java.util.Arrays;
 import java.util.List;
+import java.util.Map;
 
 public class RedisStreams101Consumer {
 
     public final static String STREAMS_KEY = "weather_sensor:wind";
+    public final static String HASH_KEY = "weather_sensor:wind:hash:";
+    public final static String MESSAGE_KEY = "weather_sensor:wind:message:";
 
 
     public static void main(String[] args) {
@@ -23,7 +24,7 @@ public static void main(String[] args) {
             syncCommands.xgroupCreate( XReadArgs.StreamOffset.from(STREAMS_KEY, "0-0"), "application_1"  );
         }
         catch (RedisBusyException redisBusyException) {
-            System.out.println( String.format("\t Group '%s' already exists", "application_1"));
+            System.out.printf("\t Group '%s' already exists%n", "application_1");
         }
 
 
@@ -41,7 +42,19 @@ public static void main(String[] args) {
                     System.out.println(message);
                     // Confirm that the message has been processed using XACK
                     syncCommands.xack(STREAMS_KEY, "application_1",  message.getId());
-                    // syncCommands.hset(message.getId(), message.getBody());
+                    Map<String, String> body = message.getBody();
+                    String hashKey = HASH_KEY + message.getId();
+                    String messageId = body.get("message_id");
+                    String messageKey = MESSAGE_KEY + messageId;
+                    String numberParts = body.get("total_parts");
+                    String thisPart = body.get("this_part");
+                    //  write a hash for each message body
+                    syncCommands.hmset(hashKey, body);
+                    //  keep track of all the hash keys for this message body
+                    syncCommands.sadd(messageKey, hashKey);
+                    if (numberParts == thisPart) {
+                        System.out.println("All Message parts received for " + messageKey);
+                    }
                 }
             }
 

src/main/java/com/jphaugla/redis/streams/RedisStreams101Consumer.java

@@ -4,24 +4,24 @@
 import io.lettuce.core.api.StatefulRedisConnection;
 import io.lettuce.core.api.sync.RedisCommands;
 
-import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Random;
 
 public class RedisStreams101Producer {
 
     public final static String STREAMS_KEY = "weather_sensor:wind";
+    public final static String MESSAGE_KEY = "MSG";
 
     public static void main(String[] args) {
 
         int nbOfMessageToSend = 1;
 
         if (args != null && args.length != 0 ) {
-            nbOfMessageToSend = Integer.valueOf(args[0]);
+            nbOfMessageToSend = Integer.parseInt(args[0]);
         }
 
-        System.out.println( String.format("\n Sending %s message(s)", nbOfMessageToSend));
+        System.out.printf("\n Sending %s message(s)%n", nbOfMessageToSend);
 
 
         RedisClient redisClient = RedisClient.create("redis://localhost:12000"); // change to reflect your environment
@@ -34,25 +34,31 @@ public static void main(String[] args) {
         for (int i = 0 ; i < nbOfMessageToSend ; i++) {
 
             Map<String, String> messageBody = new HashMap<>();
+            messageBody.put("message_id", MESSAGE_KEY + i);
+            int numberParts = new Random().nextInt(5) + 2;
+
             messageBody.put("speed", "15");
             messageBody.put("direction", "270");
             messageBody.put("sensor_ts", String.valueOf(System.currentTimeMillis()));
-            messageBody.put("loop_info", String.valueOf( i ));
-            Boolean whichone = new Random().nextBoolean();
-            String messageId = "";
-            if (whichone) {
-                messageBody.put("source_port", "12000");
-                messageId = syncCommands.xadd(
-                        STREAMS_KEY,
-                        messageBody);
-            } else {
-                messageBody.put("source_port", "12002");
-                 messageId = syncCommands2.xadd(
-                    STREAMS_KEY,
-                    messageBody);
+            for (int j = 1 ; j <= numberParts ; j++) {
+                messageBody.put("total_parts", String.valueOf(numberParts));
+                messageBody.put("this_part", String.valueOf(j));
+                boolean whichRedis = new Random().nextBoolean();
+                String messageId;
+                if (whichRedis) {
+                    messageBody.put("source_port", "12000");
+                    messageId = syncCommands.xadd(
+                            STREAMS_KEY,
+                            messageBody);
+                } else {
+                    messageBody.put("source_port", "12002");
+                    messageId = syncCommands2.xadd(
+                            STREAMS_KEY,
+                            messageBody);
+                }
+                System.out.printf("\tMessage %s : %s posted%n", messageId, messageBody);
             }
 
-            System.out.println(String.format("\tMessage %s : %s posted", messageId, messageBody));
         }
 
         System.out.println("\n");