class JedisEventConsumer extends JedisPubSub {
    static final Logger log = LoggerFactory.getLogger(JedisEventConsumer.class);

    CloudEventObserverManger observerManger;
    String eventChannelName;

    public JedisEventConsumer(CloudProps cloudProps, CloudEventObserverManger observerManger) {
        this.observerManger = observerManger;
        this.eventChannelName = cloudProps.getEventChannel();
    }

    @Override
    public void onMessage(String channel, String message) {
        try {
            Event event = ONode.deserialize(message, Event.class);
            event.channel(eventChannelName);

            onReceive(event);
        } catch (Throwable e) {
            e = Utils.throwableUnwrap(e);

            log.warn(e.getMessage(), e); //todo: ?

            if (e instanceof RuntimeException) {
                throw (RuntimeException) e;
            } else {
                throw new RuntimeException(e);
            }
        }
    }

    private boolean onReceive(Event event) {
        try {
            return onReceiveDo(event);
        } catch (Throwable e) {
            log.warn(e.getMessage(), e);
            return false;
        }
    }

    /**
     * 处理接收事件
     */
    private boolean onReceiveDo(Event event) throws Throwable {<FILL_FUNCTION_BODY>}
}

        boolean isOk = true;
        CloudEventHandler handler = null;

        //new topic
        String topicNew;
        if (Utils.isEmpty(event.group())) {
            topicNew = event.topic();
        } else {
            topicNew = event.group() + JedisProps.GROUP_SPLIT_MARK + event.topic();
        }

        handler = observerManger.getByTopic(topicNew);
        if (handler != null) {
            isOk = handler.handle(event);
        } else {
            //只需要记录一下
            log.warn("There is no observer for this event topic[{}]", topicNew);
        }

        return isOk;
    
<methods>public void <init>() <variables>