code: @Override
      public boolean subscribe(String topic, MessageListener messageListener) {
          Set<MessageListener> subcription = topicSubscriptions.putIfAbsent(topic,
                  new HashSet<MessageListener>());
          if (subcription == null) {
              subcription = topicSubscriptions.get(topic);
          }
  
          boolean subscribe = false;
          WrappedJedisPubSub wrappedJedisPubSub = null;
          synchronized (subcription) {
              if (subcription.add(messageListener)) {
                  wrappedJedisPubSub = new WrappedJedisPubSub(messageListener);
                  topicSubscriptionMappings.put(messageListener, wrappedJedisPubSub);
                  subscribe = true;
              }
          }
          if (subscribe) {
              byte[] topicName = SafeEncoder.encode(topic);
              // this operation blocks!
              redisClient.subscribe(wrappedJedisPubSub, topicName);
              return true;
          } else {
              return false;
          }
      }
docs: {@inheritDoc}