137 lines
3.8 KiB
Java
137 lines
3.8 KiB
Java
import java.io.*;
|
|
import java.util.*;
|
|
import com.erlport.erlang.term.*;
|
|
import com.erlport.*;
|
|
|
|
class State implements Serializable {
|
|
|
|
Integer times;
|
|
|
|
public State() {
|
|
times = 0;
|
|
}
|
|
|
|
public Integer incr() {
|
|
times += 1;
|
|
return times;
|
|
}
|
|
|
|
@Override
|
|
public String toString() {
|
|
return String.format("State(times: %d)", times);
|
|
}
|
|
}
|
|
|
|
public class Main {
|
|
|
|
static Integer OK = 0;
|
|
static Integer ERROR = 0;
|
|
|
|
//-------------------
|
|
// Connection level
|
|
|
|
public static Object init(Object conn, Object connInfo) {
|
|
System.err.printf("[java] established a conn=%s, connInfo=%s\n", conn, connInfo);
|
|
|
|
// set an instance to be the connection state
|
|
// it just a example structure to record the callback total times
|
|
Object state = new State();
|
|
|
|
// subscribe the topic `t/dn` with qos0
|
|
subscribe(conn, new Binary("t/dn"), 0);
|
|
|
|
// return the initial conn's state
|
|
return Tuple.two(OK, state);
|
|
}
|
|
|
|
public static Object received(Object conn, Object data, Object state) {
|
|
System.err.printf("[java] received data conn=%s, data=%s, state=%s\n", conn, data, state);
|
|
|
|
// echo the conn's data
|
|
send(conn, data);
|
|
|
|
// return the new conn's state
|
|
State nstate = (State) state;
|
|
nstate.incr();
|
|
return Tuple.two(OK, nstate);
|
|
}
|
|
|
|
public static void terminated(Object conn, Object reason, Object state) {
|
|
System.err.printf("[java] terminated conn=%s, reason=%s, state=%s\n", conn, reason, state);
|
|
return;
|
|
}
|
|
|
|
//-----------------------
|
|
// Protocol/Session level
|
|
|
|
public static Object deliver(Object conn, Object msgs0, Object state) {
|
|
System.err.printf("[java] received messages conn=%s, msgs=%s, state=%s\n", conn, msgs0, state);
|
|
|
|
List<Object> msgs = (List<Object>) msgs0;
|
|
for(Object msg: msgs) {
|
|
publish(conn, msg);
|
|
}
|
|
|
|
// return the new conn's state
|
|
State nstate = (State) state;
|
|
nstate.incr();
|
|
return Tuple.two(OK, nstate);
|
|
}
|
|
|
|
//-----------------------
|
|
// APIs
|
|
public static void send(Object conn, Object data) {
|
|
try {
|
|
Erlang.call("emqx_exproto", "send", new Object[]{conn, data}, 5000);
|
|
} catch (Exception e) {
|
|
System.err.printf("[java] send data error: %s\n", e);
|
|
}
|
|
return;
|
|
}
|
|
|
|
public static void close(Object conn) {
|
|
try {
|
|
Erlang.call("emqx_exproto", "close", new Object[]{conn}, 5000);
|
|
} catch (Exception e) {
|
|
System.err.printf("[java] send data error: %s\n", e);
|
|
}
|
|
return;
|
|
}
|
|
|
|
public static void register(Object conn, Object clientInfo) {
|
|
try {
|
|
Erlang.call("emqx_exproto", "register", new Object[]{conn, clientInfo}, 5000);
|
|
} catch (Exception e) {
|
|
System.err.printf("[java] send data error: %s\n", e);
|
|
}
|
|
return;
|
|
}
|
|
|
|
public static void publish(Object conn, Object message) {
|
|
try {
|
|
Erlang.call("emqx_exproto", "publish", new Object[]{conn, message}, 5000);
|
|
} catch (Exception e) {
|
|
System.err.printf("[java] send data error: %s\n", e);
|
|
}
|
|
return;
|
|
}
|
|
|
|
public static void subscribe(Object conn, Object topic, Object qos) {
|
|
try {
|
|
Erlang.call("emqx_exproto", "subscribe", new Object[]{conn, topic, qos}, 5000);
|
|
} catch (Exception e) {
|
|
System.err.printf("[java] send data error: %s\n", e);
|
|
}
|
|
return;
|
|
}
|
|
|
|
public static void unsubscribe(Object conn, Object topic) {
|
|
try {
|
|
Erlang.call("emqx_exproto", "unsubscribe", new Object[]{conn, topic}, 5000);
|
|
} catch (Exception e) {
|
|
System.err.printf("[java] send data error: %s\n", e);
|
|
}
|
|
return;
|
|
}
|
|
}
|