Implement listening in signal blocks.

app/src/main/java/com/codigoparallevar/minicards/CanvasView.java

@@ -66,6 +66,7 @@ public class CanvasView extends View implements PartGrid {
     @Nullable
     private Tuple2<Integer, Integer> _mouseDownPoint = null;
     private int cardBackgroundColor;
+    private SignalListenerManager listenerManager = null;
 
     public CanvasView(Context context) {
         super(context);
@@ -359,6 +360,14 @@ public class CanvasView extends View implements PartGrid {
         return api;
     }
 
+    @Override
+    public SignalListenerManager getListenerManager() {
+        if (listenerManager == null) {
+            listenerManager = new SignalListenerManager(getApi());
+        }
+        return listenerManager;
+    }
+
     @Override
     @Nullable
     public SignalInputConnector getSignalInputConnectorOn(int x, int y) {

app/src/main/java/com/codigoparallevar/minicards/SignalListenerManager.java

@@ -0,0 +1,108 @@
+package com.codigoparallevar.minicards;
+
+import android.util.Log;
+
+import com.codigoparallevar.minicards.types.functional.Tuple2;
+import com.programaker.api.ProgramakerApi;
+import com.programaker.api.ProgramakerListeningChannel;
+import com.programaker.api.ProgramakerSignalListener;
+
+import org.jetbrains.annotations.NotNull;
+
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+
+public class SignalListenerManager implements ProgramakerSignalListener {
+    private final ProgramakerApi api;
+    private final Map<Tuple2<String, String>, List<ProgramakerSignalListener>> channelToListener = new LinkedHashMap<>();
+    private final Map<ProgramakerSignalListener, List<Tuple2<String, String>>> listenerChannels = new LinkedHashMap<>();
+    private final Map<Tuple2<String, String>, ProgramakerListeningChannel> idToChannel = new LinkedHashMap<>();
+    private long RECONNECT_SLEEP_TIME = 2000; // 2seconds
+    private final static String LogTag = "Signal Listener Manager";
+
+    public SignalListenerManager(ProgramakerApi api) {
+        this.api = api;
+    }
+
+    public void registerSignalListener(String bridgeId, String key, ProgramakerSignalListener listener) {
+        Tuple2<String, String> id = new Tuple2<>(bridgeId, key);
+        if (!idToChannel.containsKey(id)) {
+            // Channel has to be opened
+            idToChannel.put(id, this.api.openChannelTo(bridgeId, key, this,
+                    () -> {
+                        SignalListenerManager.this.onDisconnect(bridgeId, key);
+                    }));
+        }
+
+        if (!channelToListener.containsKey(id)) {
+            channelToListener.put(id, new LinkedList<>());
+        }
+        List<ProgramakerSignalListener> listeners = channelToListener.get(id);
+        listeners.add(listener);
+
+        if (!listenerChannels.containsKey(listener)) {
+            listenerChannels.put(listener, new LinkedList<>());
+        }
+        listenerChannels.get(listener).add(id);
+    }
+
+    public void unregisterSignalListener(ProgramakerSignalListener listener) {
+        List<Tuple2<String, String>> channels = listenerChannels.get(listener);
+        listenerChannels.remove(listener);
+        for (Tuple2<String, String> id : channels) {
+            List<ProgramakerSignalListener> remainingListeners = channelToListener.get(id);
+
+            remainingListeners.remove(listener);
+            if (remainingListeners.size() == 0) {
+                ProgramakerListeningChannel channel = idToChannel.get(id);
+                channel.stop();
+                idToChannel.remove(id);
+                channelToListener.remove(id);
+            }
+        }
+    }
+
+    private void onDisconnect(String bridgeId, String key) {
+        Log.w(LogTag, "Connection lost to (bridge="+bridgeId+",key="+key + ")");
+        Tuple2<String, String> id = new Tuple2<>(bridgeId, key);
+
+        // On disconnect disable the connection, wait 2 seconds and retry
+        idToChannel.put(id, null);
+        
+        new Thread(() -> {
+            try {
+                Thread.sleep(RECONNECT_SLEEP_TIME);
+            } catch (InterruptedException e) {
+                e.printStackTrace();
+            }
+
+            idToChannel.put(id, this.api.openChannelTo(bridgeId, key, this,
+                    () -> {
+                        SignalListenerManager.this.onDisconnect(bridgeId, key);
+                    }));
+        }).start();
+    }
+
+    @Override
+    public void onNewSignal(@NotNull String bridgeId, @NotNull String key, @NotNull HashMap<?, ?> signal) {
+        Tuple2<String, String> id = new Tuple2<>(bridgeId, key);
+
+        if (!channelToListener.containsKey(id)) {
+            Log.e(LogTag, "Got signal to unlistened channel (bridgeId=" + bridgeId + ",key=" + key + ")");
+            return;
+        }
+
+        for (ProgramakerSignalListener listener : channelToListener.get(id)) {
+            try {
+                listener.onNewSignal(bridgeId, key, signal);
+            }
+            catch (Exception ex) {
+                Log.e(LogTag, "Error passing message (bridge=" + bridgeId + ",key" + key
+                        + ") to " + listener, ex);
+            }
+        }
+    }
+}

app/src/main/java/com/codigoparallevar/minicards/StubPartGrid.java

@@ -19,6 +19,11 @@ class StubPartGrid implements PartGrid {
         return null;
     }
 
+    @Override
+    public SignalListenerManager getListenerManager() {
+        return null;
+    }
+
     @Override
     public SignalInputConnector getSignalInputConnectorOn(int x, int y) {
         return null;

app/src/main/java/com/codigoparallevar/minicards/bridge/ProgramakerBridgeService.java

@@ -59,6 +59,7 @@ public class ProgramakerBridgeService extends Service {
                     ProgramakerBridgeService.BridgeStatusNotificationChannel,
                     ProgramakerBridgeService.BridgeStatusNotificationChannelName,
                     NotificationManager.IMPORTANCE_DEFAULT);
+            channel.enableVibration(false);
             notificationManager.createNotificationChannel(channel);
         }
 
@@ -105,7 +106,10 @@ public class ProgramakerBridgeService extends Service {
 
     @Override
     public int onStartCommand(Intent intent, int flags, int startId) {
-        String action = intent.getAction();
+        String action = null;
+        if (intent != null) { // Apparently this (intent=null) can happen...
+            action = intent.getAction();
+        }
         if (action == null) {
             action = "";
         }

app/src/main/java/com/codigoparallevar/minicards/parts/ProgramakerCustomBlockPart.java

@@ -21,21 +21,24 @@ import com.codigoparallevar.minicards.types.wireData.Signal;
 import com.codigoparallevar.minicards.types.wireData.WireDataType;
 import com.codigoparallevar.minicards.ui_helpers.DoAsync;
 import com.programaker.api.ProgramakerApi;
+import com.programaker.api.ProgramakerSignalListener;
 import com.programaker.api.data.ProgramakerCustomBlock;
 import com.programaker.api.data.ProgramakerCustomBlockArgument;
 import com.programaker.api.data.ProgramakerCustomBlockSaveTo;
 import com.programaker.api.data.ProgramakerFunctionCallResult;
 
+import org.jetbrains.annotations.NotNull;
 import org.json.JSONArray;
 import org.json.JSONException;
 import org.json.JSONObject;
 
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.UUID;
 
-public class ProgramakerCustomBlockPart implements Part {
+public class ProgramakerCustomBlockPart implements Part, ProgramakerSignalListener {
     private static final int WIDTH_PADDING = 50;
     private static final int HEIGHT_PADDING = 50;
     private static final int IO_RADIUS = 50;
@@ -59,7 +62,6 @@ public class ProgramakerCustomBlockPart implements Part {
     private Tuple2<ConnectorTypeInfo, RoundOutputConnector> saveToOutput;
     private RoundOutputConnector pulseOutput;
 
-
     public ProgramakerCustomBlockPart(String id, PartGrid grid, Tuple2<Integer, Integer> center, ProgramakerCustomBlock block) {
         this._id = id;
         this._partGrid = grid;
@@ -207,7 +209,7 @@ public class ProgramakerCustomBlockPart implements Part {
 
         this.updatePortPositions();
     }
-    
+
     private void updatePortPositions() {
         {
             // Update inputs
@@ -505,11 +507,67 @@ public class ProgramakerCustomBlockPart implements Part {
     @Override
     public void resume() {
         this.active = true;
+
+        String type = _block.getBlock_type();
+        if (type != null && (type.equals("trigger"))) {
+            // Listen to signal
+            ProgramakerApi api = _partGrid.getApi();
+            if (api == null) {
+                Log.e(LogTag, "Cannot listen to API (API not found)");
+                return;
+            }
+
+            new DoAsync().executeOnExecutor(AsyncTask.THREAD_POOL_EXECUTOR,
+                    new Tuple2<>(
+                            () -> {
+                                _partGrid.getListenerManager().registerSignalListener(
+                                        _block.getBridge_id(), _block.getKey(),
+                                        this);
+                            },
+                            ex -> {
+                                Log.e(LogTag, "Error establishing connection to monitor", ex);
+                            }
+                    ));
+            }
+    }
+
+    @Override
+    public void onNewSignal(@NotNull String bridgeId, @NotNull String key, @NotNull HashMap<?, ?> signal) {
+        // Propagate signal
+        // Stream object on save_to, then trigger pulse
+        Object content = signal.get("content");
+        if (this.saveToOutput != null) {
+            // this.saveToOutput.item2.send(content); // TODO: Have an output type that allows this
+        }
+
+        if (this.pulseOutput != null) {
+            this.pulseOutput.send(new Signal());
+        }
+        _partGrid.update();
     }
 
     @Override
     public void pause() {
         this.active = false;
+
+        String type = _block.getBlock_type();
+        if (type != null && (type.equals("trigger"))) {
+            // Release listening of signal
+            ProgramakerApi api = _partGrid.getApi();
+            if (api == null) {
+                return;
+            }
+
+            new DoAsync().executeOnExecutor(AsyncTask.THREAD_POOL_EXECUTOR,
+                    new Tuple2<>(
+                            () -> {
+                                _partGrid.getListenerManager().unregisterSignalListener(this);
+                            },
+                            ex -> {
+                                Log.e(LogTag, "Error disconnecting from monitor", ex);
+                            }
+                    ));
+        }
     }
 
     @Override
@@ -623,7 +681,8 @@ public class ProgramakerCustomBlockPart implements Part {
 
     @Override
     public void unlink() {
-        this.active = false;
+        pause();
+
         for (InputConnector input : getInputConnectors()) {
             input.unlink();
         }

app/src/main/java/com/codigoparallevar/minicards/types/PartGrid.java

@@ -1,5 +1,6 @@
 package com.codigoparallevar.minicards.types;
 
+import com.codigoparallevar.minicards.SignalListenerManager;
 import com.codigoparallevar.minicards.types.connectors.input.BooleanInputConnector;
 import com.codigoparallevar.minicards.types.connectors.input.SignalInputConnector;
 import com.codigoparallevar.minicards.types.connectors.input.StringInputConnector;
@@ -9,6 +10,7 @@ import com.programaker.api.ProgramakerApi;
 public interface PartGrid {
     Selectable getPartOn(int x, int y);
     ProgramakerApi getApi();
+    SignalListenerManager getListenerManager();
 
     SignalInputConnector getSignalInputConnectorOn(int x, int y);
     BooleanInputConnector getBooleanInputConnectorOn(int x, int y);

app/src/main/java/com/codigoparallevar/minicards/types/functional/Tuple2.java

@@ -8,4 +8,43 @@ public class Tuple2<T1, T2> {
         this.item1 = item1;
         this.item2 = item2;
     }
+
+    @Override
+    public int hashCode() {
+        int hash1 = this.item1 == null ? 1 : this.item1.hashCode();
+        int hash2 = this.item2 == null ? 2 : this.item2.hashCode();
+
+        return hash1 ^ hash2;
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (!(obj instanceof Tuple2)) {
+            return false;
+        }
+
+        Tuple2 other = (Tuple2) obj;
+        if (other.item1 == null) {
+            if (this.item1 != null) {
+                return false;
+            }
+        }
+        else {
+            if (!other.item1.equals(this.item1)) {
+                return false;
+            }
+        }
+
+        if (other.item2 == null) {
+            if (this.item2 != null) {
+                return false;
+            }
+        }
+        else {
+            if (!other.item2.equals(this.item2)) {
+                return false;
+            }
+        }
+        return true;
+    }
 }

app/src/main/java/com/programaker/api/ProgramakerApi.kt

@@ -222,6 +222,18 @@ class ProgramakerApi(private val ApiRoot: String="https://programaker.com/api")
         }
     }
 
+    fun openChannelTo(bridgeId: String, key: String,
+                      listener: ProgramakerSignalListener,
+                      onDisconnectCallback: Runnable): ProgramakerListeningChannel {
+        val channel = ProgramakerListeningChannel(
+                bridgeId, key,
+                this.token!!,
+                getListenSignalUrl(bridgeId, key),
+                listener, onDisconnectCallback)
+        channel.start()
+        return channel
+    }
+
     // Private functions
     private fun getCheckUrl(): String {
         return "$ApiRoot/v0/sessions/check"
@@ -261,6 +273,11 @@ class ProgramakerApi(private val ApiRoot: String="https://programaker.com/api")
         return "$ApiRoot/v0/users/$userName/services/id/$bridgeId/register"
     }
 
+    private fun getListenSignalUrl(bridgeId: String, key: String): String {
+        this.withUserId()
+        return "$ApiRoot/v0/users/id/$userId/bridges/id/$bridgeId/signals/$key"
+    }
+
     fun getUserId(): String? {
         this.withUserId()
         return userId;

app/src/main/java/com/programaker/api/ProgramakerListeningChannel.kt

@@ -0,0 +1,78 @@
+package com.programaker.api
+
+import android.util.Log
+import com.google.gson.Gson
+import okhttp3.*
+import okio.ByteString
+import org.json.JSONException
+import java.nio.charset.Charset
+import java.util.concurrent.TimeUnit
+
+class ProgramakerListeningChannel(
+        private val bridgeId: String,
+        private val key: String,
+        private val token: String,
+        private val url: String,
+        private val listener: ProgramakerSignalListener,
+        private var onDisconnect: Runnable
+): WebSocketListener() {
+    private var webSocket: WebSocket? = null
+    private val utf8: Charset = Charset.forName("UTF-8")
+    private val gson = Gson()
+
+    private val LogTag: String = "PM-ListeningChannel"
+    private val PING_PERIOD_MILLIS: Long = 15000 // 15seconds
+
+    fun start() {
+        val client: OkHttpClient = OkHttpClient.Builder()
+                .pingInterval(PING_PERIOD_MILLIS, TimeUnit.MILLISECONDS)
+                .readTimeout(0, TimeUnit.MILLISECONDS)
+                .build()
+
+        val request: Request = Request.Builder()
+                .url(url)
+                .addHeader("Authorization", token)
+                .build()
+        client.newWebSocket(request, this)
+
+        // Trigger shutdown of the dispatcher's executor so this process can exit cleanly.
+        client.dispatcher.executorService.shutdown()
+    }
+
+    // Websocket management
+    override fun onOpen(webSocket: WebSocket, response: Response) {
+        this.webSocket = webSocket
+    }
+
+    override fun onMessage(webSocket: WebSocket, bytes: ByteString) {
+        onMessage(webSocket, bytes.string(utf8))
+    }
+
+    override fun onMessage(webSocket: WebSocket, text: String) {
+        Log.d(LogTag, "Message: $text")
+        val json = gson.fromJson(text, HashMap::class.java)
+        if (json == null) {
+            this.onFailure(webSocket, JSONException("Error decoding: $text"), null)
+        }
+        else {
+            this.listener.onNewSignal(bridgeId, key, json)
+        }
+    }
+
+    override fun onClosing(webSocket: WebSocket, code: Int, reason: String) {
+        webSocket.close(1000, null)
+        Log.i(LogTag, "Closing bridge socket {code=$code, reason=$reason}")
+        onDisconnect.run()
+    }
+
+    override fun onFailure(webSocket: WebSocket, t: Throwable, response: Response?) {
+        Log.e(LogTag, "Error: $t", t)
+        onDisconnect.run()
+    }
+
+    fun stop() {
+        onDisconnect = Runnable {  } // Skip disconnection procedure
+        webSocket?.close(1000, null)
+        webSocket = null
+    }
+}

app/src/main/java/com/programaker/api/ProgramakerSignalListener.kt

@@ -0,0 +1,7 @@
+package com.programaker.api
+
+import java.util.HashMap
+
+interface ProgramakerSignalListener {
+    fun onNewSignal(bridgeId: String, key: String, signal: HashMap<*, *>)
+}

app/src/main/java/com/programaker/api/data/ProgramakerCustomBlock.kt

@@ -11,7 +11,9 @@ class ProgramakerCustomBlock(
         val block_type: String?,
         val block_result_type: String?,
         var bridge_id: String?,
-        val save_to: ProgramakerCustomBlockSaveTo?
+        val save_to: ProgramakerCustomBlockSaveTo?,
+        var key: String?,
+        val subkey: ProgramakerCustomBlockSubkeyDefinition?
 ) {
     fun serialize(): JSONObject {
 
@@ -33,10 +35,17 @@ class ProgramakerCustomBlock(
                 "message" to message,
                 "arguments" to serializedArguments,
                 "block_type" to block_type,
-                "block_result_type" to block_result_type,
                 "bridge_id" to bridge_id,
                 "save_to" to saveToSerialized
-        );
+        )
+
+        if (key != null) {
+            serialized.put("key", key)
+            serialized.put("subkey", subkey?.serialize())
+        }
+        if (block_result_type != null || key == null) {
+            serialized.put("block_result_type", block_result_type)
+        }
 
         return JSONObject(serialized as Map<*, *>)
     }
@@ -49,9 +58,11 @@ class ProgramakerCustomBlock(
                     obj.getString("message"),
                     ProgramakerCustomBlockArgument.deserialize(obj.optJSONArray("arguments")),
                     obj.getString("block_type"),
-                    obj.getString("block_result_type"),
+                    obj.optString("block_result_type"),
                     obj.optString("bridge_id"),
-                    ProgramakerCustomBlockSaveTo.deserialize(obj.optJSONObject("save_to"))
+                    ProgramakerCustomBlockSaveTo.deserialize(obj.optJSONObject("save_to")),
+                    obj.optString("key"),
+                    ProgramakerCustomBlockSubkeyDefinition.deserialize(obj.optJSONObject("subkey"))
             )
 
             return block

app/src/main/java/com/programaker/api/data/ProgramakerCustomBlockSaveTo.kt

@@ -24,5 +24,4 @@ class ProgramakerCustomBlockSaveTo (
                     save_to.getInt("index"))
         }
     }
-
 }

app/src/main/java/com/programaker/api/data/ProgramakerCustomBlockSubkeyDefinition.kt

@@ -0,0 +1,29 @@
+package com.programaker.api.data
+
+import org.json.JSONObject
+
+class ProgramakerCustomBlockSubkeyDefinition(
+        val type: String,
+        val value: String
+) {
+    fun serialize(): JSONObject {
+        return JSONObject(hashMapOf(
+                "type" to type,
+                "value" to value
+        ) as Map<*, *>)
+    }
+
+    companion object {
+        @JvmStatic
+        fun deserialize(subkey: JSONObject?): ProgramakerCustomBlockSubkeyDefinition? {
+            if (subkey == null) {
+                return null;
+            }
+
+            return ProgramakerCustomBlockSubkeyDefinition(
+                    subkey.getString("type"),
+                    subkey.getString("value"))
+        }
+    }
+
+}