package JNode;

import com.ericsson.otp.erlang.*;
import java.io.IOException;


/**
 * Created by wangchunye on 3/13/16.
 */
public class Node {
    Producer producer;
    public static void main (String[]args) {
        new Thread(new Runnable() {
            public void run() {
                try {
                    System.out.println("start to read from standard input");
                    int buf = System.in.read();
                    System.out.println("receive:" + buf);
                } catch (IOException e) {
                    System.exit(0);
                }
            }}).start();
        Node node = new Node();
        try {
            node.loop();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    Node() {
        producer = new Producer();
    }
    void loop() throws IOException {
        String FirstNode = System.getenv("FIRST_NODE");
        if(FirstNode == null){
            System.err.println("env FIRST_NODE is empty");
            System.exit(1);
        }
        OtpNode self;
        String Cookie = System.getenv("ERLANG_COOKIE");
        if(Cookie == null) {
            self = new OtpNode("java");
        } else {
            self = new OtpNode("java", Cookie);
        }
        OtpMbox msgBox = self.createMbox("kafka");
        if (!self.ping(FirstNode, 2000)) {
            System.err.println("unable to connect the first node " + FirstNode);
            System.exit(1);
        }
        System.out.println("java node is created.");
        OtpErlangObject exit = new OtpErlangAtom("exit");
        while (true) {
            try {
                OtpErlangObject o = msgBox.receive();
                System.out.println("Hello " + o + "   " + o.equals(exit));
                if (o instanceof OtpErlangTuple) {
                    OtpErlangTuple msg = (OtpErlangTuple)o;

                    OtpErlangPid from = (OtpErlangPid)(msg.elementAt(0));
                    OtpErlangObject ref = msg.elementAt(1);
                    OtpErlangBinary topic = (OtpErlangBinary)msg.elementAt(2);
                    OtpErlangBinary data = (OtpErlangBinary)msg.elementAt(3);
                    this.producer.send(new String(topic.binaryValue()), new String(data.binaryValue()));

                    OtpErlangObject[] result = new OtpErlangObject[2];
                    result[0] = ref;
                    result[1] = new OtpErlangAtom("ok");
                    msgBox.send(from, new OtpErlangTuple(result));
                }else if( o.equals(exit) ) {
                    System.exit(0);
                }
            } catch (OtpErlangExit otpErlangExit) {
                otpErlangExit.printStackTrace();
            } catch (OtpErlangDecodeException e) {
                e.printStackTrace();
            }
        }
    }
}

@@ -44,85 +44,424 @@
 ang.
-*;%0Aimport java.io.IOException;%0A%0A%0A/**%0A * Created by wangchunye on 3/13/16.%0A */
+OtpErlangAtom;%0Aimport com.ericsson.otp.erlang.OtpErlangBinary;%0Aimport com.ericsson.otp.erlang.OtpErlangDecodeException;%0Aimport com.ericsson.otp.erlang.OtpErlangExit;%0Aimport com.ericsson.otp.erlang.OtpErlangObject;%0Aimport com.ericsson.otp.erlang.OtpErlangPid;%0Aimport com.ericsson.otp.erlang.OtpErlangTuple;%0Aimport com.ericsson.otp.erlang.OtpMbox;%0Aimport com.ericsson.otp.erlang.OtpNode;%0A%0Aimport java.io.IOException;%0A%0A
 %0Apub
@@ -647,32 +647,34 @@
                 
+//
 System.out.print
@@ -2015,83 +2015,8 @@
 ();%0A
-                System.out.println(%22Hello %22 + o + %22   %22 + o.equals(exit));%0A
     

remove debugging code