Pass sourceNodeId to messengers

Author: s1ck

pregel/src/main/java/org/neo4j/gds/beta/pregel/AsyncQueueMessenger.java

@@ -41,7 +41,7 @@ public void initIteration(int iteration) {
     }
 
     @Override
-    public void sendTo(long targetNodeId, double message) {
+    public void sendTo(long sourceNodeId, long targetNodeId, double message) {
         assert !Double.isNaN(message);
         queues.push(targetNodeId, message);
     }

pregel/src/main/java/org/neo4j/gds/beta/pregel/Messenger.java

@@ -23,7 +23,7 @@ public interface Messenger<ITERATOR extends Messages.MessageIterator> {
 
     void initIteration(int iteration);
 
-    void sendTo(long targetNodeId, double message);
+    void sendTo(long sourceNodeId, long targetNodeId, double message);
 
     ITERATOR messageIterator();
 

pregel/src/main/java/org/neo4j/gds/beta/pregel/ReducingMessenger.java

@@ -77,7 +77,7 @@ public void initIteration(int iteration) {
     }
 
     @Override
-    public void sendTo(long targetNodeId, double message) {
+    public void sendTo(long sourceNodeId, long targetNodeId, double message) {
         sendArray.update(
             targetNodeId,
             current -> reducer.reduce(current, message)

pregel/src/main/java/org/neo4j/gds/beta/pregel/SyncQueueMessenger.java

@@ -39,7 +39,7 @@ public void initIteration(int iteration) {
     }
 
     @Override
-    public void sendTo(long targetNodeId, double message) {
+    public void sendTo(long sourceNodeId, long targetNodeId, double message) {
         queues.push(targetNodeId, message);
     }
 

pregel/src/main/java/org/neo4j/gds/beta/pregel/context/ComputeContext.java

@@ -153,7 +153,7 @@ public void sendToNeighbors(double message) {
      * @throws ArrayIndexOutOfBoundsException if the node is in the not in id space
      */
     public void sendTo(long targetNodeId, double message) {
-        messenger.sendTo(targetNodeId, message);
+        messenger.sendTo(this.nodeId, targetNodeId, message);
         this.hasSendMessage.setValue(true);
     }