Merge pull request #8 from skodjob/redis

Add Redis endpoint and fix parent paths

database-manipulation-tool-schema/pom.xml

@@ -14,6 +14,7 @@
     <artifactId>database-performance-hub</artifactId>
     <groupId>io.skodjob</groupId>
     <version>1.0.0</version>
+    <relativePath>../pom.xml</relativePath>
   </parent>
 
   <properties>

database-manipulation-tool/README.md

@@ -23,6 +23,8 @@ When using DMT for load generation in performance testing, use the `performance`
 
 This is currently in alpha state so there are not many features completed and the ones that are finished aren't tested properly.
 
+### Databases
+
 | Database/feature        	       | Postgres 	 | Mongo 	 | Mysql 	 | Oracle 	 | SqlServer 	 | Db2 	 |
 |---------------------------------|------------|---------|---------|----------|-------------|-------|
 | Create table            	       | X    	     | X   	   | X   	   | 	        | 	           | 	     |
@@ -32,6 +34,18 @@ This is currently in alpha state so there are not many features completed and th
 | Drop table        	             | X	         | X	      | X	      | 	        | 	           | 	     |
 | Reset Database        	         | X	         | 	X      | X	      | 	        | 	           | 	     |
 
+### Streams
+| Streaming system | Redis | Pulsar | RabbitMQ |
+|------------------|-------|--------|----------|
+| Poll from stream | X     |        |          |
+| Send to stream   | X     |        |          |
+| Auth             |       |        |          |
+
+### Redis configuration
+- `data.source.redis.host` -- Redis host (`localhost`)
+- `data.source.redis.port` -- Redis port (`6379`)
+- `data.source.redis.pool.max` -- Amount of connections open to Redis (`10`)
+
 ## Json insertion schema
 The Json schema is `DatabaseEntry` class from the DMT-schema project.  
 ```
@@ -80,6 +94,30 @@ You can reset all databases on DMT start with property `onstart.reset.database`.
 <br />
 <summary><code>POST</code> <code><b>/Main/GenerateLoad?count={count}&maxRows={maxRows}</b></code> <code>(Creates set number of queries with maximum table size and upserts them into databases. Returns the time length of the whole request and just the execution of the queries)</code></summary>
 <summary><code>POST</code> <code><b>/Main/GenerateBatchLoad?count={count}&maxRows={maxRows}</b></code> <code>(Same as GenerateLoad but uses batch statements for better performance)</code></summary>
+<br />
+<summary>
+    <code>GET</code> 
+    <code><b>/Redis/pollMessages?max=5</b></code> 
+    <code>
+        [
+            "channel1",
+            "channel2"
+        ]
+    </code>
+    <code>Reads at most 'max' messages from specified channels</code>
+</summary>
+<summary>
+    <code>GET</code> 
+    <code><b>/Redis/sendMessage?channel=channel1</b></code> 
+    <code>
+        {
+            "rider": "Alonso",
+            "position": 1
+        }
+    </code>
+    <code>Sends a message specified in json body to the redis stream channel 
+    specified in as header argument</code>
+</summary>
 
 ## Running the application in dev mode
 

database-manipulation-tool/pom.xml

@@ -9,6 +9,7 @@
     <artifactId>database-performance-hub</artifactId>
     <groupId>io.skodjob</groupId>
     <version>1.0.0</version>
+    <relativePath>../pom.xml</relativePath>
   </parent>
 
   <properties>
@@ -84,7 +85,11 @@
       <groupId>io.quarkus</groupId>
       <artifactId>quarkus-micrometer-registry-prometheus</artifactId>
     </dependency>
-
+    <dependency>
+      <groupId>redis.clients</groupId>
+      <artifactId>jedis</artifactId>
+      <version>4.3.1</version>
+    </dependency>
     <!-- Test dependencies -->
     <dependency>
       <groupId>io.quarkus</groupId>

database-manipulation-tool/src/main/java/io/skodjob/dmt/resource/RedisResource.java

@@ -0,0 +1,44 @@
+package io.skodjob.dmt.resource;
+
+import java.util.List;
+import java.util.Map;
+
+import io.quarkus.runtime.annotations.RegisterForReflection;
+import io.skodjob.dmt.service.RedisService;
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.inject.Inject;
+import jakarta.ws.rs.Consumes;
+import jakarta.ws.rs.GET;
+import jakarta.ws.rs.POST;
+import jakarta.ws.rs.Path;
+import jakarta.ws.rs.Produces;
+import jakarta.ws.rs.QueryParam;
+import jakarta.ws.rs.core.MediaType;
+import jakarta.ws.rs.core.Response;
+import redis.clients.jedis.resps.StreamEntry;
+
+@Path("Redis")
+@Produces(MediaType.APPLICATION_JSON)
+@Consumes(MediaType.APPLICATION_JSON)
+@ApplicationScoped
+@RegisterForReflection
+public class RedisResource {
+
+    @Inject
+    RedisService redisService;
+
+    @Path("sendMessage")
+    @POST
+    public Response sendMessage(@QueryParam("channel") String channel, Map<String, String> message) {
+        String newId = redisService.insert(channel, message);
+        return Response.ok(newId).build();
+    }
+
+    @Path("pollMessages")
+    @GET
+    public Response pollMessages(@QueryParam("max") Integer max, List<String> channels) {
+        List<Map.Entry<String, List<StreamEntry>>> res = redisService.get(max, channels);
+        return Response.ok(res).build();
+    }
+
+}

database-manipulation-tool/src/main/java/io/skodjob/dmt/service/RedisService.java

@@ -0,0 +1,23 @@
+package io.skodjob.dmt.service;
+
+import java.util.List;
+import java.util.Map;
+
+import io.skodjob.dmt.sinkSources.RedisDataSource;
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.inject.Inject;
+import redis.clients.jedis.resps.StreamEntry;
+
+@ApplicationScoped
+public class RedisService {
+    @Inject
+    RedisDataSource redisDataSource;
+
+    public String insert(String channel, Map<String, String> message) {
+        return redisDataSource.sendMessage(channel, message).toString();
+    }
+
+    public List<Map.Entry<String, List<StreamEntry>>> get(int amount, List<String> channels) {
+        return redisDataSource.pollMessages(amount, channels);
+    }
+}

database-manipulation-tool/src/main/java/io/skodjob/dmt/sinkSources/RedisDataSource.java

@@ -0,0 +1,50 @@
+package io.skodjob.dmt.sinkSources;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.inject.Inject;
+import org.eclipse.microprofile.config.inject.ConfigProperty;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import redis.clients.jedis.Jedis;
+import redis.clients.jedis.JedisPool;
+import redis.clients.jedis.JedisPoolConfig;
+import redis.clients.jedis.StreamEntryID;
+import redis.clients.jedis.params.XAddParams;
+import redis.clients.jedis.params.XReadParams;
+import redis.clients.jedis.resps.StreamEntry;
+
+@ApplicationScoped
+public class RedisDataSource {
+
+    protected final Logger LOGGER = LoggerFactory.getLogger(this.getClass());
+
+    private JedisPool pool;
+
+    @Inject
+    public RedisDataSource(@ConfigProperty(name = "data.source.redis.host") String host,
+                           @ConfigProperty(name = "data.source.redis.port") Integer port,
+                           @ConfigProperty(name = "data.source.redis.pool.max") Integer maxPool) {
+        JedisPoolConfig config = new JedisPoolConfig();
+        config.setMaxTotal(maxPool);
+        this.pool = new JedisPool(config, host, port);
+    }
+
+    public StreamEntryID sendMessage(String channel, Map<String, String> message) {
+        Jedis jedis = pool.getResource();
+        return jedis.xadd(channel, message, XAddParams.xAddParams());
+    }
+
+    public List<Map.Entry<String, List<StreamEntry>>> pollMessages(int maxAmount, List<String> channels) {
+
+        Jedis jedis = pool.getResource();
+        HashMap<String, StreamEntryID> streams = new HashMap<>();
+        channels.forEach(channel -> streams.put(channel, new StreamEntryID()));
+
+        return jedis.xread(XReadParams.xReadParams().count(maxAmount), streams);
+    }
+
+}

database-manipulation-tool/src/main/resources/application.properties

@@ -38,4 +38,6 @@ quarkus.vertx.max-event-loop-execute-time=1000
 
 %classic.Retry/maxRetries=5
 
-
+data.source.redis.host=localhost
+data.source.redis.port=6379
+data.source.redis.pool.max=10

load-generator/pom.xml

@@ -13,6 +13,7 @@
     <artifactId>database-performance-hub</artifactId>
     <groupId>io.skodjob</groupId>
     <version>1.0.0</version>
+    <relativePath>../pom.xml</relativePath>
   </parent>
 
   <properties>