@Override
    public int addListener(ObjectListener listener) {
        if (listener instanceof StreamAddListener) {
            return addListener("__keyevent@*:xadd", (StreamAddListener) listener, StreamAddListener::onAdd);
        }
        if (listener instanceof StreamRemoveListener) {
            return addListener("__keyevent@*:xdel", (StreamRemoveListener) listener, StreamRemoveListener::onRemove);
        }
        if (listener instanceof StreamCreateConsumerListener) {
            return addListener("__keyevent@*:xgroup-createconsumer", (StreamCreateConsumerListener) listener, StreamCreateConsumerListener::onCreateConsumer);
        }
        if (listener instanceof StreamRemoveConsumerListener) {
            return addListener("__keyevent@*:xgroup-delconsumer", (StreamRemoveConsumerListener) listener, StreamRemoveConsumerListener::onRemoveConsumer);
        }
        if (listener instanceof StreamCreateGroupListener) {
            return addListener("__keyevent@*:xgroup-create", (StreamCreateGroupListener) listener, StreamCreateGroupListener::onCreateGroup);
        }
        if (listener instanceof StreamRemoveGroupListener) {
            return addListener("__keyevent@*:xgroup-destroy", (StreamRemoveGroupListener) listener, StreamRemoveGroupListener::onRemoveGroup);
        }
        if (listener instanceof StreamTrimListener) {
            return addListener("__keyevent@*:xtrim", (StreamTrimListener) listener, StreamTrimListener::onTrim);
        }
        if (listener instanceof TrackingListener) {
            return addTrackingListener((TrackingListener) listener);
        }

        return super.addListener(listener);
    }
@Test
    public void testAddListener() {
        testWithParams(redisson -> {
            RStream<String, String> ss = redisson.getStream("test");
            ss.createGroup(StreamCreateGroupArgs.name("test-group").makeStream());
            CountDownLatch latch = new CountDownLatch(1);
            ss.addListener(new StreamAddListener() {
                @Override
                public void onAdd(String name) {
                    latch.countDown();
                }
            });
            ss.add(StreamAddArgs.entry("test1", "test2"));

            try {
                assertThat(latch.await(1, TimeUnit.SECONDS)).isTrue();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }, NOTIFY_KEYSPACE_EVENTS, "Et");
    }