Extend RabbitMQ into the Web using PubNub

1.js

{"Depositor":"Randy","Amount":123.00}

2.js

 public void successCallback(String channelPubNub, Object message) {
        System.out.println(" [x] SubscribeAdapter : SUBSCRIBE : RECEIVED on channel:" + channelPubNub
                + " : " + message.getClass() + " : "
                + message.toString());
        Produce(message, channelRabbitMQ);
    }

3.js

try {
            //for this demo, we will produce a message to RabbitMQ onbly if we received a valid number in the Amount element of the JSON message from PubNub
            JSONObject jsobj = new JSONObject(message.toString());
            if (jsobj.getDouble("Amount") >= 0 ){
                System.out.println(" [*] SubscribeAdapter : PRODUCE : verified Amount: " + jsobj.getDouble("Amount"));
                jsobj.append("Verifier", "SubscribeAdapter");
                channelRabbitMQ.basicPublish( "", TASK_QUEUE_NAME,
                        MessageProperties.PERSISTENT_TEXT_PLAIN,
                        jsobj.toString().getBytes());
                System.out.println(" [+] SubscribeAdapter : PRODUCE : produced to " + 
                        TASK_QUEUE_NAME + " '" + jsobj.toString() + "'");
            }
            else{
                System.out.println(" [-] SubscribeAdapter : PRODUCE : discarding due to invalid Amount in " + message);
            }
        } catch (Exception e) {
            System.out.println(" [!] SubscribeAdapter : PRODUCE : discarding message due to exception: " + e);
        }

4.js

{"Amount":123,"Verifier":["SubscribeAdapter"],"Depositor":"Randy"}

5.js

while (true) {
    QueueingConsumer.Delivery delivery = consumer.nextDelivery();
    String message = new String(delivery.getBody());    
    System.out.println(" [x] PublishAdapter : received '" + message + "'");
    try {
        JSONObject jsobj = new JSONObject(message);
        //execute some business logic to decide whether or not to publish to PubNub
        if ( jsobj.getDouble("Amount") > threshold )
        {
            System.out.println(" [*] PublishAdapter : forwarding, Amount (" + jsobj.getDouble("Amount") + ") exceeds threshold (" + threshold + ") so let's publish to PubNub");
            jsobj.put("Threshold", threshold);
            jsobj.append("Verifier", "PublishAdapter");
            pubnub.publish(channelName, jsobj, new Callback(){
                public void successCallback(String channel, Object message) {
                    System.out.println(" [+] PublishAdapter : published to PubNub " + message);
                }
                public void errorCallback(String channel, PubnubError error) {
                    System.out.println(" [!] PublishAdapter : error " + error);
                }
            });
        }
        else
            System.out.println(" [-] PublishAdapter : discarding, Amount (" + jsobj.getDouble("Amount") + ") is below threshold (" + threshold + ") so let's not publish to PubNub");
    }catch (JSONException e) {
        System.out.println(" [!] PublishAdapter : error " + e);
    }
    channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
}