@Override
    public void run() {
        boolean forward = false;
        Object message;
        Subscription subscription = null;
        try {
            message = this.ois.readObject();
            if (message instanceof Message) {
                Message msg = (Message) message;
                matchSubscriptions(msg);
            } else if (message instanceof Subscription) {
                subscription = (Subscription) message;
                System.out.println("Subscription from " + subscription.getIpAddress());
                forward = addSubscription(subscription);
            }

        } catch (IOException | ClassNotFoundException ex) {
            Logger.getLogger(BrokerNodeHandler.class.getName()).log(Level.SEVERE, null, ex);
        }
        if (forward) {
            String hostAddress;
            hostAddress = this.socket.getInetAddress().getHostAddress();
            subscription.setFromIpAddress(hostAddress);
            forwardMessage(subscription);
        }

    }