Support list patterns

.ci/docker-compose.override.yml

@@ -0,0 +1,5 @@
+version: '3'
+
+services:
+    logstash:
+      network_mode: host

.ci/run.sh

@@ -0,0 +1,10 @@
+#!/bin/bash
+# This is intended to be run inside the docker container as the command of the docker-compose.
+
+env
+
+set -ex
+
+jruby -rbundler/setup -S rspec -fd
+
+jruby -rbundler/setup -S rspec -fd --tag redis

.travis.yml

@@ -1,2 +1,13 @@
 import:
-- logstash-plugins/.ci:travis/[email protected]
+- logstash-plugins/.ci:travis/[email protected]
+
+addons:
+  apt:
+    sources:
+      - sourceline: 'ppa:chris-lea/redis-server'
+    packages:
+      - redis-server
+
+before_install:
+  - sudo service redis-server stop
+  - sudo service redis-server start --bind 0.0.0.0

CHANGELOG.md

@@ -1,3 +1,13 @@
+## 3.7.0
+  - Fix: better (Redis) exception handling [#89](https://github.com/logstash-plugins/logstash-input-redis/pull/89)
+  - Test: start running integration specs on CI
+
+## 3.6.1
+  - Fix: resolve crash when commands_map is set [#86](https://github.com/logstash-plugins/logstash-input-redis/pull/86)
+
+## 3.6.0
+  - Remove ruby pipeline dependency. Starting from Logstash 8, Ruby execution engine is not available. All pipelines should use Java pipeline [#84](https://github.com/logstash-plugins/logstash-input-redis/pull/84)
+
 ## 3.5.1
   - [DOC] Reordered config option to alpha order [#79](https://github.com/logstash-plugins/logstash-input-redis/issues/79)
 

README.md

@@ -1,7 +1,7 @@
 # Logstash Plugin
 
 Travis Build
-[![Travis Build Status](https://travis-ci.org/logstash-plugins/logstash-input-redis.svg)](https://travis-ci.org/logstash-plugins/logstash-input-redis)
+[![Travis Build Status](https://travis-ci.com/logstash-plugins/logstash-input-redis.svg)](https://travis-ci.com/logstash-plugins/logstash-input-redis)
 
 This is a plugin for [Logstash](https://github.com/elastic/logstash).
 

batch_perf/perf_batch.rb

@@ -3,7 +3,7 @@
 require "securerandom"
 
 require "logstash/event"
-require "logstash/pipeline"
+require "logstash/java_pipeline"
 require_relative "../lib/logstash/inputs/redis"
 
 class BenchOptions
@@ -35,7 +35,7 @@ def cfg_batch(d)
 bench_options = BenchOptions.new
 
 def input(cfg, slow, &block)
-  pipeline = LogStash::Pipeline.new(cfg)
+  pipeline = LogStash::JavaPipeline.new(cfg)
   queue = Queue.new
 
   pipeline.instance_eval do

docs/index.asciidoc

@@ -84,12 +84,14 @@ Redis allows for the renaming or disabling of commands in its protocol, see:  ht
 ===== `data_type`
 
   * This is a required setting.
-  * Value can be any of: `list`, `channel`, `pattern_channel`
+  * Value can be any of: `list`, `pattern_list`, `channel`, `pattern_channel`
   * There is no default value for this setting.
 
 Specify either list or channel.  If `data_type` is `list`, then we will BLPOP the
-key.  If `data_type` is `channel`, then we will SUBSCRIBE to the key.
-If `data_type` is `pattern_channel`, then we will PSUBSCRIBE to the key.
+key. If `data_type` is `pattern_list`, then we will spawn a number of worker
+threads that will LPOP from keys matching that pattern.  If `data_type` is
+`channel`, then we will SUBSCRIBE to the key. If `data_type` is
+`pattern_channel`, then we will PSUBSCRIBE to the key.
 
 [id="plugins-{type}s-{plugin}-db"]
 ===== `db`
@@ -125,6 +127,7 @@ The unix socket path of your Redis server.
 
 The name of a Redis list or channel.
 
+
 [id="plugins-{type}s-{plugin}-password"]
 ===== `password`
 
@@ -133,6 +136,37 @@ The name of a Redis list or channel.
 
 Password to authenticate with. There is no authentication by default.
 
+
+[id="plugins-{type}s-{plugin}-pattern_list_max_items"]
+===== `pattern_list_max_items`
+
+  * Value type is <<number,number>>
+  * Default value is `1000`
+
+Maximum number of items for a single worker thread to process when `data_type` is `pattern_list`.
+After the list is empty or this number of items have been processed, the thread will exit and a
+new one will be started if there are non-empty lists matching the pattern without a consumer.
+
+
+[id="plugins-{type}s-{plugin}-pattern_list_threadpool_sleep"]
+===== `pattern_list_threadpool_sleep`
+
+  * Value type is <<number,number>>
+  * Default value is `0.2`
+
+Time to sleep in main loop after checking if more threads can/need to be spawned.
+Applies to `data_type` is `pattern_list`
+
+
+[id="plugins-{type}s-{plugin}-pattern_list_threads"]
+===== `pattern_list_threads`
+
+  * Value type is <<number,number>>
+  * Default value is `20`
+
+Maximum number of worker threads to spawn when using `data_type` `pattern_list`.
+
+
 [id="plugins-{type}s-{plugin}-port"]
 ===== `port`
 
@@ -141,8 +175,9 @@ Password to authenticate with. There is no authentication by default.
 
 The port to connect on.
 
+
 [id="plugins-{type}s-{plugin}-ssl"]
-===== `ssl` 
+===== `ssl`
 
   * Value type is <<boolean,boolean>>
   * Default value is `false`
@@ -157,7 +192,6 @@ Enable SSL support.
   * Default value is `1`
 
 
-
 [id="plugins-{type}s-{plugin}-timeout"]
 ===== `timeout`
 
@@ -166,7 +200,9 @@ Enable SSL support.
 
 Initial connection timeout in seconds.
 
+
+
 [id="plugins-{type}s-{plugin}-common-options"]
 include::{include_path}/{type}.asciidoc[]
 
-:default_codec!:
+:default_codec!: