@Override
    public void sendMessage(final MessageScope messageScope, final M message) {
        if (messageScope instanceof MessageScope.Local) {
            final MessageScope.Local<M> localMessageScope = (MessageScope.Local) messageScope;
            final Traversal.Admin<Vertex, Edge> incidentTraversal = SparkMessenger.setVertexStart(localMessageScope.getIncidentTraversal().get().asAdmin(), this.vertex);
            final Direction direction = SparkMessenger.getOppositeDirection(incidentTraversal);

            // handle processing for BOTH given TINKERPOP-1862 where the target of the message is the one opposite
            // the current vertex
            incidentTraversal.forEachRemaining(edge -> {
                if (direction.equals(Direction.IN) || direction.equals(Direction.OUT))
                    this.outgoingMessages.add(new Tuple2<>(edge.vertices(direction).next().id(), localMessageScope.getEdgeFunction().apply(message, edge)));
                else
                    this.outgoingMessages.add(new Tuple2<>(edge instanceof StarGraph.StarOutEdge ? edge.inVertex().id() : edge.outVertex().id(), localMessageScope.getEdgeFunction().apply(message, edge)));

            });
        } else {
            ((MessageScope.Global) messageScope).vertices().forEach(v -> this.outgoingMessages.add(new Tuple2<>(v.id(), message)));
        }
    }
@author Marko A. Rodriguez (http://markorodriguez.com)
spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkMessenger.java