class ActivemqConsumeHandler implements MessageListener {
    static final Logger log = LoggerFactory.getLogger(ActivemqConsumeHandler.class);

    private CloudEventObserverManger observerManger;
    private Session session;

    public ActivemqConsumeHandler(CloudEventObserverManger observerManger, Session session) {
        super();
        this.observerManger = observerManger;
        this.session = session;
    }

    @Override
    public void onMessage(Message message) {
        ActiveMQTextMessage textmsg = (ActiveMQTextMessage) message;
        try {
            Event event = ONode.deserialize(textmsg.getText(), Event.class);
            event.times(textmsg.getRedeliveryCounter());

            //已设置自动延时策略
            boolean isOk = onReceive(event);

            if(isOk){
                textmsg.acknowledge();
            }else{
                session.recover();
            }
        } 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) throws Throwable {
        try {
            return onReceiveDo(event);
        } catch (Throwable e) {
            log.warn(e.getMessage(), e);
            return false;
        }
    }

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

    private String getTopicNew(Event event){
        return ActivemqProps.getTopicNew(event);
    }
}

        boolean isOk = true;
        CloudEventHandler handler = null;

        //new topic
        String topicNew = getTopicNew(event);

        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;
    
<no_super_class>