@Override
    public void accept(Metadata metadata, InputStream content) throws IOException {
        // handling of service type messages
        if (metadata.getType() == Metadata.Type.SERVICE) {
            List<MessageReceiver> receivers = new ArrayList<>();
            synchronized (consumers) {
                Map<String, MessageReceiver> topicConsumers = this.consumers.row(metadata.getTopic());
                // we need to filter receivers if the message is targeted only
                if (metadata.hasTargets()) {
                    for (String target : metadata.getTargets()) {
                        MessageReceiver receiver = topicConsumers.get(target);
                        if (receiver != null) {
                            receivers.add(receiver);
                        }
                    }
                } else {
                    receivers.addAll(topicConsumers.values());
                }
            }
            // call all receivers
            for (MessageReceiver receiver : receivers) {
                receiver.accept(metadata, content);
            }
        }
        // handling of callbacks
        if (metadata.getType() == Metadata.Type.CALLBACK) {
            MessageReceiver receiver;
            synchronized (consumers) {
                receiver = this.consumers.get(Metadata.CALLBACK_TOPIC, metadata.getReplyTo());
            }
            if (receiver != null) {
                receiver.accept(metadata, content);
            } else {
                LOG.warn("No receiver for callback with replyTo {}", metadata.getReplyTo());
            }
        }
    }