@Override
    public void execute(final Vertex vertex, Messenger<Pair<Serializable, Double>> messenger, final Memory memory) {
        if (memory.isInitialIteration()) {
            if (this.distributeVote) {
                messenger.sendMessage(this.countScope, Pair.with('c', 1.0d));
            } else {
                double voteStrength = (null == this.initialVoteStrengthTraversal ?
                        1.0d :
                        TraversalUtil.apply(vertex, this.initialVoteStrengthTraversal.get()).doubleValue());
                vertex.property(VertexProperty.Cardinality.single, this.property, vertex.id());
                vertex.property(VertexProperty.Cardinality.single, VOTE_STRENGTH, voteStrength);
                messenger.sendMessage(this.voteScope, new Pair<>((Serializable) vertex.id(), voteStrength));
                memory.add(VOTE_TO_HALT, false);
            }
        } else if (1 == memory.getIteration() && this.distributeVote) {
            double voteStrength = (null == this.initialVoteStrengthTraversal ?
                    1.0d :
                    TraversalUtil.apply(vertex, this.initialVoteStrengthTraversal.get()).doubleValue()) /
                    IteratorUtils.reduce(IteratorUtils.map(messenger.receiveMessages(), Pair::getValue1), 0.0d, (a, b) -> a + b);
            vertex.property(VertexProperty.Cardinality.single, this.property, vertex.id());
            vertex.property(VertexProperty.Cardinality.single, VOTE_STRENGTH, voteStrength);
            messenger.sendMessage(this.voteScope, new Pair<>((Serializable) vertex.id(), voteStrength));
            memory.add(VOTE_TO_HALT, false);
        } else {
            final Map<Serializable, Double> votes = new HashMap<>();
            votes.put(vertex.value(this.property), vertex.<Double>value(VOTE_STRENGTH));
            messenger.receiveMessages().forEachRemaining(message -> MapHelper.incr(votes, message.getValue0(), message.getValue1()));
            Serializable cluster = PeerPressureVertexProgram.largestCount(votes);
            if (null == cluster) cluster = (Serializable) vertex.id();
            memory.add(VOTE_TO_HALT, vertex.value(this.property).equals(cluster));
            vertex.property(VertexProperty.Cardinality.single, this.property, cluster);
            messenger.sendMessage(this.voteScope, new Pair<>(cluster, vertex.<Double>value(VOTE_STRENGTH)));
        }
    }
@author Marko A. Rodriguez (http://markorodriguez.com)
gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/computer/clustering/peerpressure/PeerPressureVertexProgram.java