<commit_before>package com.openxc.remote;

import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;

import java.net.URISyntaxException;
import java.net.URI;

import java.util.ArrayList;

import java.util.concurrent.CopyOnWriteArrayList;

import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import com.google.common.base.Objects;

import com.openxc.remote.sinks.DataSinkException;
import com.openxc.remote.sinks.VehicleDataSink;

import com.openxc.remote.sources.DataSourceException;
import com.openxc.remote.sources.SourceCallback;
import com.openxc.remote.sources.VehicleDataSource;

import android.util.Log;

/**
 * The DataPipeline ferries raw messages from VehicleDataSources to
 * VehicleDataSinks.
 */
public class DataPipeline implements SourceCallback {
    private final static String TAG = "DataPipeline";

    private int mMessagesReceived = 0;
    // TODO move all of the measurement stuff to a sink
    // and have the RVS create and handle that sink itself
    private Map<String, RawMeasurement> mMeasurements;
    private CopyOnWriteArrayList<VehicleDataSink> mSinks;
    private CopyOnWriteArrayList<VehicleDataSource> mSources;

    public DataPipeline() {
        mMeasurements = new HashMap<String, RawMeasurement>();
        mSinks = new CopyOnWriteArrayList<VehicleDataSink>();
        mSources = new CopyOnWriteArrayList<VehicleDataSource>();
    }

    public void receive(String measurementId, Object value, Object event) {
        for(Iterator<VehicleDataSink> i = mSinks.iterator(); i.hasNext();) {
            (i.next()).receive(measurementId, value, event);
        }
        mMessagesReceived++;
    }

    public void removeSink(VehicleDataSink sink) {
        if(sink != null) {
            mSinks.remove(sink);
            sink.stop();
        }
    }

    public void removeSink(String sinkName) {
        removeEndpoint(mSinks, sinkName);
    }

    public VehicleDataSink addSink(VehicleDataSink sink) {
        mSinks.add(sink);
        return sink;
    }


    public List<VehicleDataSink> getSinks() {
        return mSinks;
    }

    public List<VehicleDataSource> getSources() {
        return mSources;
    }

    // TODO we need to set ourselves as the callback for the source?
    public void addSource(VehicleDataSource source) {
        if(source != null) {
            Log.i(TAG, "Initializing vehicle data source " + source);
            new Thread(source).start();
        }

        mSources.add(source);
    }

    public void removeSource(VehicleDataSource source) {
        if(source != null) {
            mSources.remove(source);
            source.stop();
        }
    }

    public void removeSource(String sourceName) {
        removeEndpoint(mSources, sourceName);
    }<commit_msg>Remove android.util.Log usage from DataPipeline to avoid "Stub!"<commit_after>package com.openxc.remote;

import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;

import java.net.URISyntaxException;
import java.net.URI;

import java.util.ArrayList;

import java.util.concurrent.CopyOnWriteArrayList;

import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import com.google.common.base.Objects;

import com.openxc.remote.sinks.DataSinkException;
import com.openxc.remote.sinks.VehicleDataSink;

import com.openxc.remote.sources.DataSourceException;
import com.openxc.remote.sources.SourceCallback;
import com.openxc.remote.sources.VehicleDataSource;

/**
 * The DataPipeline ferries raw messages from VehicleDataSources to
 * VehicleDataSinks.
 */
public class DataPipeline implements SourceCallback {
    private final static String TAG = "DataPipeline";

    private int mMessagesReceived = 0;
    // TODO move all of the measurement stuff to a sink
    // and have the RVS create and handle that sink itself
    private Map<String, RawMeasurement> mMeasurements;
    private CopyOnWriteArrayList<VehicleDataSink> mSinks;
    private CopyOnWriteArrayList<VehicleDataSource> mSources;

    public DataPipeline() {
        mMeasurements = new HashMap<String, RawMeasurement>();
        mSinks = new CopyOnWriteArrayList<VehicleDataSink>();
        mSources = new CopyOnWriteArrayList<VehicleDataSource>();
    }

    public void receive(String measurementId, Object value, Object event) {
        for(Iterator<VehicleDataSink> i = mSinks.iterator(); i.hasNext();) {
            (i.next()).receive(measurementId, value, event);
        }
        mMessagesReceived++;
    }

    public void removeSink(VehicleDataSink sink) {
        if(sink != null) {
            mSinks.remove(sink);
            sink.stop();
        }
    }

    public void removeSink(String sinkName) {
        removeEndpoint(mSinks, sinkName);
    }

    public VehicleDataSink addSink(VehicleDataSink sink) {
        mSinks.add(sink);
        return sink;
    }


    public List<VehicleDataSink> getSinks() {
        return mSinks;
    }

    public List<VehicleDataSource> getSources() {
        return mSources;
    }

    // TODO we need to set ourselves as the callback for the source?
    public void addSource(VehicleDataSource source) {
        if(source != null) {
            new Thread(source).start();
        }

        mSources.add(source);
    }

    public void removeSource(VehicleDataSource source) {
        if(source != null) {
            mSources.remove(source);
            source.stop();
        }
    }

    public void removeSource(String sourceName) {
        removeEndpoint(mSources, sourceName);
    }
{'fraction_non_alphanumeric': 0.09352132786007496, 'fraction_numerical': 0.0003569516330537212, 'mean_word_length': 3.1572700296735907, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 21, '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': '133812', 'n_tokens_mistral': 1590, 'n_tokens_neox': 1605, 'n_words': 379}