<commit_before>package com.i906.mpt.location;

import android.content.Context;
import android.location.Location;
import android.os.Bundle;
import android.support.annotation.NonNull;
import android.support.annotation.Nullable;

import com.google.android.gms.common.ConnectionResult;
import com.google.android.gms.common.api.GoogleApiClient;
import com.google.android.gms.location.LocationListener;
import com.google.android.gms.location.LocationRequest;
import com.google.android.gms.location.LocationServices;

import java.util.ArrayList;
import java.util.List;

import rx.AsyncEmitter;
import rx.Observable;
import rx.functions.Action1;

/**
 * @author <NAME>
 */
class RxFusedLocation {

    private GoogleApiClient mClient;
    private ApiCallbacks mCallbacks;

    RxFusedLocation(Context context) {
        mCallbacks = new ApiCallbacks();
        mClient = new GoogleApiClient.Builder(context)
                .addApi(LocationServices.API)
                .addConnectionCallbacks(mCallbacks)
                .addOnConnectionFailedListener(mCallbacks)
                .build();
    }

    public void connect() {
        if (!mClient.isConnected()) {
            mClient.connect();
        }
    }

    public Observable<Location> getLocation(LocationRequest request) {
        return Observable.fromEmitter(new LocationUpdates(request),
                AsyncEmitter.BackpressureMode.LATEST);
    }

    public void disconnect() {
        if (mClient.isConnected() || mClient.isConnecting()) {
            mClient.disconnect();
        }
    }

    private class LocationUpdates implements Action1<AsyncEmitter<Location>>,
            AsyncEmitter.Cancellable,
            LocationListener {

        private final LocationRequest request;
        private AsyncEmitter<Location> emitter;

        LocationUpdates(LocationRequest request) {
            this.request = request;
        }

        @Override
        public void call(AsyncEmitter<Location> emitter) {
            this.emitter = emitter;
            this.emitter.setCancellation(this);
            mCallbacks.addObserver(this);

            if (mClient.isConnected()) {
                requestLocationUpdates();
            }
        }

        private void requestLocationUpdates() {
            try {
                LocationServices.FusedLocationApi.requestLocationUpdates(mClient, request, this);
            } catch (SecurityException e) {
                this.emitter.onError(e);
            }
        }

        @Override
        public void onLocationChanged(Location location) {
            emitter.onNext(location);
        }

        private void onConnected() {
            if (emitter != null) {
                requestLocationUpdates();
            }
        }

        private void onError(Throwable e) {
            if (emitter != null) {
                emitter.onError(e);
            }
        }

        @Override
        public void cancel() throws Exception {
            LocationServices.FusedLocationApi.removeLocationUpdates(mClient, this);
            mCallbacks.removeCallback(this);
        }
    }

    private class ApiCallbacks implements GoogleApiClient.ConnectionCallbacks,
            GoogleApiClient.OnConnectionFailedListener {

        private List<LocationUpdates> observers;

        ApiCallbacks() {
            observers = new ArrayList<>();
        }

        void addObserver(LocationUpdates observer) {
            if (!observers.contains(observer)) observers.add(observer);
            connect();
        }

        void removeCallback(LocationUpdates observer) {
            observers.remove(observer);

            if (observers.isEmpty()) {
                disconnect();
            }
        }

        @Override
        public void onConnected(@Nullable Bundle bundle) {<commit_msg>Fix concurrent exception<commit_after>package com.i906.mpt.location;

import android.content.Context;
import android.location.Location;
import android.os.Bundle;
import android.support.annotation.NonNull;
import android.support.annotation.Nullable;

import com.google.android.gms.common.ConnectionResult;
import com.google.android.gms.common.api.GoogleApiClient;
import com.google.android.gms.location.LocationListener;
import com.google.android.gms.location.LocationRequest;
import com.google.android.gms.location.LocationServices;

import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;

import rx.AsyncEmitter;
import rx.Observable;
import rx.functions.Action1;

/**
 * @author <NAME>
 */
class RxFusedLocation {

    private GoogleApiClient mClient;
    private ApiCallbacks mCallbacks;

    RxFusedLocation(Context context) {
        mCallbacks = new ApiCallbacks();
        mClient = new GoogleApiClient.Builder(context)
                .addApi(LocationServices.API)
                .addConnectionCallbacks(mCallbacks)
                .addOnConnectionFailedListener(mCallbacks)
                .build();
    }

    public void connect() {
        if (!mClient.isConnected()) {
            mClient.connect();
        }
    }

    public Observable<Location> getLocation(LocationRequest request) {
        return Observable.fromEmitter(new LocationUpdates(request),
                AsyncEmitter.BackpressureMode.LATEST);
    }

    public void disconnect() {
        if (mClient.isConnected() || mClient.isConnecting()) {
            mClient.disconnect();
        }
    }

    private class LocationUpdates implements Action1<AsyncEmitter<Location>>,
            AsyncEmitter.Cancellable,
            LocationListener {

        private final LocationRequest request;
        private AsyncEmitter<Location> emitter;

        LocationUpdates(LocationRequest request) {
            this.request = request;
        }

        @Override
        public void call(AsyncEmitter<Location> emitter) {
            this.emitter = emitter;
            this.emitter.setCancellation(this);
            mCallbacks.addObserver(this);

            if (mClient.isConnected()) {
                requestLocationUpdates();
            }
        }

        private void requestLocationUpdates() {
            try {
                LocationServices.FusedLocationApi.requestLocationUpdates(mClient, request, this);
            } catch (SecurityException e) {
                this.emitter.onError(e);
            }
        }

        @Override
        public void onLocationChanged(Location location) {
            emitter.onNext(location);
        }

        private void onConnected() {
            if (emitter != null) {
                requestLocationUpdates();
            }
        }

        private void onError(Throwable e) {
            if (emitter != null) {
                emitter.onError(e);
            }
        }

        @Override
        public void cancel() throws Exception {
            LocationServices.FusedLocationApi.removeLocationUpdates(mClient, this);
            mCallbacks.removeCallback(this);
        }
    }

    private class ApiCallbacks implements GoogleApiClient.ConnectionCallbacks,
            GoogleApiClient.OnConnectionFailedListener {

        private List<LocationUpdates> observers;

        ApiCallbacks() {
            observers = new CopyOnWriteArrayList<>();
        }

        void addObserver(LocationUpdates observer) {
            if (!observers.contains(observer)) observers.add(observer);
            connect();
        }

        void removeCallback(LocationUpdates observer) {
            observers.remove(observer);

            if (observers.isEmpty()) {
                disconnect();
            }
        }

        @Override
        public void onConnected(@Nullable Bundle bundle) {
{'fraction_non_alphanumeric': 0.08806595132164355, 'fraction_numerical': 0.0013085579691180318, 'mean_word_length': 2.2167508417508417, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 19, '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': '4738927', 'n_tokens_mistral': 1980, 'n_tokens_neox': 1900, 'n_words': 360}