private void consumer(String[] subscriptions) {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        try {
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();

            channel.exchangeDeclare(EXCHANGE_NAME, "direct");
            String queueName = channel.queueDeclare().getQueue();

            // all consumers subscribe to the control flow
            channel.queueBind(queueName, EXCHANGE_NAME, RK_CONTROL);

            // custom subscriptions
            for(String subscription: subscriptions) {
                boolean matching = false;
                for(Severity sev: Severity.values()) {
                    if(subscription.equalsIgnoreCase(sev.name())) {
                        channel.queueBind(queueName, EXCHANGE_NAME, sev.name());
                        System.out.println("Subscribing to " + sev.name() + " messages");
                        matching = true;
                        break;
                    }
                }
                if(!matching)
                    System.out.println(subscription + " severity not available");
            }

            System.out.println("Waiting for messages.");

            QueueingConsumer consumer = new QueueingConsumer(channel);
            channel.basicConsume(queueName, true, consumer);

            while (true) {
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                String key = delivery.getEnvelope().getRoutingKey();
                byte[] body = delivery.getBody();
                if(key.compareTo(RK_CONTROL) == 0 && body.length == 0) {
                    System.out.println("Control terminator detected.");
                    break;
                }
                String message = new String(body);

                System.out.println("Routing key: " + key + ", message: " + message);
            }

            channel.close();
            connection.close();
        }
        catch(IOException|InterruptedException e) {
            e.printStackTrace();
        }
    }