<commit_before>package com.tinkerpop.blueprints.tinkergraph;

import com.tinkerpop.blueprints.Edge;
import com.tinkerpop.blueprints.Vertex;
import com.tinkerpop.blueprints.computer.MessageType;
import com.tinkerpop.blueprints.computer.Messenger;
import com.tinkerpop.blueprints.util.StreamFactory;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;

/**
 * @author <NAME> (http://markorodriguez.com)
 */
public class TinkerMessenger<M extends Serializable> implements Messenger<M> {

    public Map<Object, Map<String, Queue<M>>> sendMessages = new HashMap<>();
    public Map<Object, Map<String, Queue<M>>> receiveMessages = new HashMap<>();

    public Iterable<M> receiveMessages(final Vertex vertex, final MessageType messageType) {
        if (messageType instanceof MessageType.Adjacent) {
            MessageType.Adjacent<Object, M> adjacentMessageType = (MessageType.Adjacent) messageType;

            final List<Edge> edge = new ArrayList<>(1); // simulates storage side-effects available in Gremlin, but not Java8 streams
            return StreamFactory.iterable(StreamFactory.stream(adjacentMessageType.getQuery().build().reverse().build(vertex).edges())
                    .map(e -> {
                        edge.clear();
                        edge.add(e);
                        return receiveMessages.get(e.getVertex(adjacentMessageType.getQuery().direction).getId());
                    })
                    .filter(m -> null != m)
                    .map(m -> m.get(messageType.getLabel()))
                    .filter(l -> null != l)
                    .flatMap(l -> l.stream())
                    .map(message -> adjacentMessageType.getEdgeFunction().apply(edge.get(0), message)));

        } else {
            return StreamFactory.iterable(Arrays.asList(vertex).stream()
                    .map(v -> this.receiveMessages.get(v.getId()))
                    .filter(m -> null != m)
                    .map(m -> m.get(messageType.getLabel()))
                    .filter(l -> null != l)
                    .flatMap(l -> l.stream()));
        }
    }

    public void sendMessage(final Vertex vertex, final MessageType messageType, final M message) {
        if (messageType instanceof MessageType.Adjacent) {
            getMessageList(vertex.getId(), messageType).add(message);
        } else {
            ((MessageType.Direct) messageType).vertices().forEach(v -> {
                getMessageList(v.getId(), messageType).add(message);
            });
        }
    }

    private Queue<M> getMessageList(final Object vertexId, final MessageType messageType) {
        Map<String, Queue<M>> messages = this.sendMessages.get(vertexId);
        if (null == messages) {
            messages = new HashMap<>();
            this.sendMessages.put(vertexId, messages);
        }
        Queue<M> messageList = messages.get(messageType.getLabel());
        if (null == messageList) {
            messageList = new ConcurrentLinkedQueue<>();
            messages.put(messageType.getLabel(), messageList);
        }
        return messageList;
    }

    public void completeIteration() {
        this.receiveMessages = this.sendMessages;
        this.sendMessages = new HashMap<>();
    }
}
<commit_msg>changed from generic function to edgeFunction.<commit_after>package com.tinkerpop.blueprints.tinkergraph;

import com.tinkerpop.blueprints.Edge;
import com.tinkerpop.blueprints.Vertex;
import com.tinkerpop.blueprints.computer.MessageType;
import com.tinkerpop.blueprints.computer.Messenger;
import com.tinkerpop.blueprints.util.StreamFactory;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;

/**
 * @author <NAME> (http://markorodriguez.com)
 */
public class TinkerMessenger<M extends Serializable> implements Messenger<M> {

    // Map<VertexId, Map<MessageLabel, MessageQueue>>>
    public Map<Object, Map<String, Queue<M>>> sendMessages = new HashMap<>();
    public Map<Object, Map<String, Queue<M>>> receiveMessages = new HashMap<>();

    public Iterable<M> receiveMessages(final Vertex vertex, final MessageType messageType) {
        if (messageType instanceof MessageType.Adjacent) {
            MessageType.Adjacent<Object, M> adjacentMessageType = (MessageType.Adjacent) messageType;

            final List<Edge> edge = new ArrayList<>(1); // simulates storage side-effects available in Gremlin, but not Java8 streams
            return StreamFactory.iterable(StreamFactory.stream(adjacentMessageType.getQuery().build().reverse().build(vertex).edges())
                    .map(e -> {
                        edge.clear();
                        edge.add(e);
                        return receiveMessages.get(e.getVertex(adjacentMessageType.getQuery().direction).getId());
                    })
                    .filter(m -> null != m)
                    .map(m -> m.get(messageType.getLabel()))
                    .filter(l -> null != l)
                    .flatMap(l -> l.stream())
                    .map(message -> adjacentMessageType.getEdgeFunction().apply(edge.get(0), message)));

        } else {
            return StreamFactory.iterable(Arrays.asList(vertex).stream()
                    .map(v -> this.receiveMessages.get(v.getId()))
                    .filter(m -> null != m)
                    .map(m -> m.get(messageType.getLabel()))
                    .filter(l -> null != l)
                    .flatMap(l -> l.stream()));
        }
    }

    public void sendMessage(final Vertex vertex, final MessageType messageType, final M message) {
        if (messageType instanceof MessageType.Adjacent) {
            getMessageList(vertex.getId(), messageType).add(message);
        } else {
            ((MessageType.Direct) messageType).vertices().forEach(v -> {
                getMessageList(v.getId(), messageType).add(message);
            });
        }
    }

    private Queue<M> getMessageList(final Object vertexId, final MessageType messageType) {
        Map<String, Queue<M>> messages = this.sendMessages.get(vertexId);
        if (null == messages) {
            messages = new HashMap<>();
            this.sendMessages.put(vertexId, messages);
        }
        Queue<M> messageList = messages.get(messageType.getLabel());
        if (null == messageList) {
            messageList = new ConcurrentLinkedQueue<>();
            messages.put(messageType.getLabel(), messageList);
        }
        return messageList;
    }

    public void completeIteration() {
        this.receiveMessages = this.sendMessages;
        this.sendMessages = new HashMap<>();
    }
}

{'fraction_non_alphanumeric': 0.12427578215527231, 'fraction_numerical': 0.0008690614136732329, 'mean_word_length': 2.450774612693653, 'pattern_counts': {'":': 0, '<': 49, '<?xml version=': 0, '>': 74, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '265312', 'n_tokens_mistral': 1899, 'n_tokens_neox': 1832, 'n_words': 372}