added zeromq listeners and async tasks

ZDmitry · Dec 26, 2016 · 82ad46a · 82ad46a
1 parent e9d34bd
commit 82ad46a
Show file tree

Hide file tree

Showing 7 changed files with 154 additions and 38 deletions.
diff --git a/android/src/main/AndroidManifest.xml b/android/src/main/AndroidManifest.xml
@@ -1,3 +1,5 @@
 <manifest xmlns:android="http://schemas.android.com/apk/res/android" package="org.zeromq.rnzeromq">
 
+    <uses-permission android:name="android.permission.INTERNET" />
+
 </manifest>
diff --git a/android/src/main/java/org/zeromq/rnzeromq/EventEmitter.java b/android/src/main/java/org/zeromq/rnzeromq/EventEmitter.java
@@ -2,6 +2,7 @@
 
 import android.util.Log;
 
+import com.facebook.react.bridge.Arguments;
 import com.facebook.react.bridge.ReactContext;
 import com.facebook.react.bridge.WritableMap;
 import com.facebook.react.modules.core.DeviceEventManagerModule;
@@ -20,9 +21,12 @@ private EventEmitter(ReactContext reactContext) {
         this.context.addLifecycleEventListener(ReceiverHelper.getInstance(context));
     }
 
-    public void emit(String eventName, WritableMap message) {
+    public void emit(final String eventName, final String message) {
         if (context.hasActiveCatalystInstance() && this.context.hasCurrentActivity()) {
-            context.getJSModule(DeviceEventManagerModule.RCTDeviceEventEmitter.class).emit(eventName, message);
+            WritableMap retVal = Arguments.createMap();
+            retVal.putString("result", message);
+
+            context.getJSModule(DeviceEventManagerModule.RCTDeviceEventEmitter.class).emit(eventName, retVal);
         }
     }
 

diff --git a/android/src/main/java/org/zeromq/rnzeromq/MessageHandler.java b/android/src/main/java/org/zeromq/rnzeromq/MessageHandler.java
@@ -0,0 +1,14 @@
+package org.zeromq.rnzeromq;
+
+import android.os.Handler;
+import android.os.Message;
+
+class MessageHandler extends Handler {
+
+    private final EventEmitter emitter = EventEmitter.getInstance();
+
+    @Override
+    public void handleMessage(Message msg) {
+        emitter.emit("zeromq.recvd", (String)msg.obj);
+    }
+}
diff --git a/android/src/main/java/org/zeromq/rnzeromq/ReactNativeZeroMQ.java b/android/src/main/java/org/zeromq/rnzeromq/ReactNativeZeroMQ.java
@@ -1,5 +1,8 @@
 package org.zeromq.rnzeromq;
 
+import android.os.Handler;
+import android.os.Message;
+
 import java.util.HashMap;
 import java.lang.String;
 import java.lang.Boolean;
@@ -12,51 +15,39 @@
 import com.facebook.react.bridge.Callback;
 import com.facebook.react.bridge.ReadableMap;
 
-import zmq.ZMQ;
-import zmq.Ctx;
-import zmq.SocketBase;
+import android.os.AsyncTask;
+
 
 class ReactNativeZeroMQ extends ReactContextBaseJavaModule {
 
-    private Ctx        _zmqCtx = null;
-    private SocketBase _dealer = null;
+    private String serverAddress = "";
+    private Thread zeroMQThread;
 
     ReactNativeZeroMQ(final ReactApplicationContext reactContext) {
         super(reactContext);
     }
 
-    private Boolean _init(final ReadableMap config) throws Exception {
-        Boolean success;
-
-        if (_zmqCtx == null) {
-            _zmqCtx = ZMQ.init(1);
-            SocketBase dealer = ZMQ.socket(_zmqCtx, ZMQ.ZMQ_DEALER);
-            if (dealer != null) {
-                String devIdentifier = "android.os.Build." + ReactNativeUtils.getDeviceName() + " " + ReactNativeUtils.getIPAddress(true);
-                ZMQ.setSocketOption(dealer, ZMQ.ZMQ_IDENTITY, devIdentifier);
-                success = ZMQ.connect(dealer, config.getString("server"));
-                if (success) {
-                    _dealer = dealer;
-                    return true;
-                }
-            }
+    private Boolean _init(final ReadableMap config) {
+        if (zeroMQThread == null) {
+            serverAddress = config.getString("server");
+            zeroMQThread  = new Thread(new ZeroMQServer(serverAddress));
+            zeroMQThread.start();
+            return true;
         }
-
         return false;
     }
 
-    private Boolean _destroy() throws Exception {
-        if (_dealer != null) {
-            ZMQ.close(_dealer);
-            _dealer = null;
-        }
+    private String _getDeviceIdentifier() {
+        return ("android.os.Build." + ReactNativeUtils.getDeviceName() + " " + ReactNativeUtils.getIPAddress(true));
+    }
 
-        if (_zmqCtx != null) {
-            ZMQ.term(_zmqCtx);
-            _zmqCtx = null;
+    private Boolean _disconnect() throws Exception {
+        if (zeroMQThread != null) {
+            zeroMQThread.interrupt();
+            zeroMQThread = null;
+            return true;
         }
-
-        return true;
+        return false;
     }
 
     @Override
@@ -66,26 +57,54 @@ public String getName() {
 
     @ReactMethod
     @SuppressWarnings("unused")
-    public void start(final ReadableMap config, Callback callback) {
+    public void connect(final ReadableMap config, final Callback callback) {
         (new ReactTask(callback) {
             @Override
             Object run() throws Exception {
-                ReactNativeZeroMQ.this._init(config);
-                return this._successResult(true);
+                Boolean success = ReactNativeZeroMQ.this._init(config);
+                return this._successResult(success);
             }
         }).start();
     }
 
     @ReactMethod
     @SuppressWarnings("unused")
-    public void destroy(Callback callback) {
+    public void sendMessage(final String message, final Callback callback) {
+        if (serverAddress.length() > 0) {
+            new ZeroMQTask(serverAddress, new Handler() {
+                @Override
+                public void handleMessage(final Message msg) {
+                    (new ReactTask(callback) {
+                        @Override
+                        Object run() throws Exception {
+                            return msg.obj;
+                        }
+                    }).start();
+                }
+            }).executeOnExecutor(AsyncTask.THREAD_POOL_EXECUTOR, message);
+        }
+    }
+
+    @ReactMethod
+    @SuppressWarnings("unused")
+    public void disconnect(final Callback callback) {
         (new ReactTask(callback) {
             @Override
             Object run() throws Exception {
-                Boolean success = ReactNativeZeroMQ.this._destroy();
+                Boolean success = ReactNativeZeroMQ.this._disconnect();
                 return this._successResult(success);
             }
         }).start();
     }
 
+    @ReactMethod
+    @SuppressWarnings("unused")
+    public void getDeviceIdentifier(final Callback callback) {
+        (new ReactTask(callback) {
+            @Override
+            Object run() throws Exception {
+                return ReactNativeZeroMQ.this._getDeviceIdentifier();
+            }
+        }).start();
+    }
 }
diff --git a/android/src/main/java/org/zeromq/rnzeromq/ReceiverHelper.java b/android/src/main/java/org/zeromq/rnzeromq/ReceiverHelper.java
@@ -9,6 +9,7 @@
 class ReceiverHelper implements LifecycleEventListener {
 
     private static ReceiverHelper instance = null;
+
     private Context context;
     private Intent  intent;
 

diff --git a/android/src/main/java/org/zeromq/rnzeromq/ZeroMQServer.java b/android/src/main/java/org/zeromq/rnzeromq/ZeroMQServer.java
@@ -0,0 +1,34 @@
+package org.zeromq.rnzeromq;
+
+import android.os.Handler;
+
+import org.zeromq.ZMQ;
+
+
+class ZeroMQServer implements Runnable {
+
+    private final String  _serverAddress;
+    private final Handler _threadHandler = new MessageHandler();
+
+    ZeroMQServer(final String serverAddr) {
+        _serverAddress = serverAddr;
+    }
+
+    @Override
+    public void run() {
+        ZMQ.Context context = ZMQ.context(1);
+        ZMQ.Socket   socket = context.socket(ZMQ.REP);
+
+        socket.bind(_serverAddress);
+        while(!Thread.currentThread().isInterrupted()) {
+            byte[] msg = socket.recv(0);
+            if (msg.length > 0) {
+                _threadHandler.sendMessage(_threadHandler.obtainMessage(0, new String(msg)));
+                socket.send("", 0);
+            }
+        }
+
+        socket.close();
+        context.term();
+    }
+}
diff --git a/android/src/main/java/org/zeromq/rnzeromq/ZeroMQTask.java b/android/src/main/java/org/zeromq/rnzeromq/ZeroMQTask.java
@@ -0,0 +1,42 @@
+package org.zeromq.rnzeromq;
+
+import android.os.AsyncTask;
+import android.os.Handler;
+
+import org.zeromq.ZMQ;
+
+
+class ZeroMQTask extends AsyncTask<String, Void, String> {
+
+    private final String  _serverAddress;
+    private       Handler _threadHandler;
+
+    ZeroMQTask(final String serverAddress, final Handler handler) {
+        _serverAddress = serverAddress;
+        _threadHandler = handler;
+
+        if (_threadHandler == null) {
+            _threadHandler = new MessageHandler();
+        }
+    }
+
+    @Override
+    protected String doInBackground(String... params) {
+        ZMQ.Context context = ZMQ.context(1);
+        ZMQ.Socket socket = context.socket(ZMQ.REQ);
+        socket.connect(_serverAddress);
+
+        socket.send(params[0].getBytes(), 0);
+        String result = new String(socket.recv(0));
+
+        socket.close();
+        context.term();
+
+        return result;
+    }
+
+    @Override
+    protected void onPostExecute(String result) {
+        _threadHandler.sendMessage(_threadHandler.obtainMessage(0, result));
+    }
+}