producer/consumer example based on wait/notify

producer_consumer.java

package com.mobidat.wp2.missionserviceImpl;

import android.graphics.Bitmap;
import android.graphics.BitmapFactory;
import android.location.Location;
import android.support.annotation.Nullable;
import android.util.Log;

import com.mobidat.dao.IDaoMission;
import com.mobidat.dao.IDaoMissionData;
import com.mobidat.dao.impl.DaoMission;
import com.mobidat.dao.impl.DaoMissionData;
import com.mobidat.dao.impl.DaoMissionDataInput;
import com.mobidat.dao.impl.DaoMissionDataMerit;
import com.mobidat.dao.impl.DaoMissionEquipment;
import com.mobidat.persistence.Activity;
import com.mobidat.persistence.Employee;
import com.mobidat.persistence.Merit;
import com.mobidat.persistence.Mission;
import com.mobidat.persistence.MissionData;
import com.mobidat.persistence.MissionDataInput;
import com.mobidat.persistence.MissionDataMerit;
import com.mobidat.persistence.MissionEquipment;
import com.mobidat.persistence.MissionState;
import com.mobidat.persistence.Profile;
import com.mobidat.persistence.Session;
import com.mobidat.persistence.Tour;
import com.mobidat.persistence.enums.AnalogNumber;
import com.mobidat.persistence.enums.DigitalNumber;
import com.mobidat.persistence.enums.InputNumber;
import com.mobidat.wp2.compatibility.GpsData;
import com.mobidat.wp2.databaseaccess.DbHelper;
import com.mobidat.wp2.gpsProvider.GPSInfo;
import com.mobidat.wp2.gpsProvider.ILocationReceiver;
import com.mobidat.wp2.gpsProvider.filter.ILocationFilter;
import com.mobidat.wp2.gpsProvider.filter.SimpleLocationFilter;
import com.mobidat.wp2.spreaderReceiver.IMobidatDataReceiver;
import com.mobidat.wp2.spreaderReceiver.InputData;
import com.mobidat.wp2.spreaderReceiver.SpreaderData;

import java.io.ByteArrayOutputStream;
import java.io.File;
import java.util.Date;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.commons.lang3.builder.EqualsBuilder;
import org.apache.log4j.Logger;

public class MissionManager implements Runnable, ILocationReceiver, IMobidatDataReceiver {

    private static final String TAG = MissionManager.class.getSimpleName();

    static final long MAX_TIME_DRIFT = 2000; // [ms]
    static final long MAX_POSITION_AGE = 2000; // [ms]
    Logger logger = Logger.getLogger(this.getClass());
    private static final MissionManager instance = new MissionManager();
    private Thread thread = null;
    private final ILocationFilter locationFilter = new SimpleLocationFilter();
    Mission mission = null;
    AtomicBoolean missionActive = new AtomicBoolean(false);
    long startTime = 0;
    long endTime = 0;
    long timeOffset = 0;
    Location location = null;
    Activity activity;
    String activityImage = null;
    private final int photoTargetWidth = 1280;
    String activityComment = null;
    Double activityQuantity = null;
    //Order order;
    Tour tour;
    Employee codriver;
    private SpreaderData spreaderData;
    private InputData inputData;
    Profile profile;
    Integer streetstateNum;
    Integer weatherNum;
    static final IDaoMission daoMission = DaoMission.getInstance();
    static final IDaoMissionData daoMissionData = DaoMissionData.getInstance();
    Merit[] merits;
    boolean equipmentsSet = false;
    private EqualsBuilder equalsBuilder = new EqualsBuilder();

    private MissionManager() {
        reset();
    }

    public static MissionManager getInstance() {
        return instance;
    }

    public void finishActiveMissions() {
        for (Mission m : daoMission.getAll()) {
            if (m.getState() == MissionState.ACTIVE) {
                m.setState(MissionState.FINISHED);
                daoMission.update(m);
                logger.warn("finish active mission with id " + m.getId());
            }
        }
    }

    public synchronized void startMission() {
        if (missionActive.compareAndSet(false, true)) {
            startTime = new Date().getTime();
            mission = new Mission();
            mission.setState(MissionState.ACTIVE);
            setMissionEmployeeIfNotSet(3000);
            mission.setProfile(profile);
            mission.setTour(tour);
            mission.setCodriver(codriver);
            activity = null;
            activityComment=null;
            activityImage=null;
            //order = null;
            daoMission.insert(mission);
            if (equipmentsSet){
                setEquipments();
            }

            thread = new Thread(this);
            thread.setName(this.getClass().getSimpleName());
            thread.start();

            logger.info("start mission with id " + mission.getId());
        } else {
            logger.error("can't start new mission, another mission is active");
        }
    }
    public synchronized void stopMission() {
        if (missionActive.compareAndSet(true, false)) {
            endTime = new Date().getTime();
            setMissionEmployeeIfNotSet(5000);
            notify();
            reset();
            mission.setState(MissionState.FINISHED);
            daoMission.update(mission);
            logger.info("stop mission");
        } else {
            logger.error("no active mission");
        }
    }

    private synchronized void reset() {
        location = null;
        activity = null;
        activityComment = null;
        activityImage=null;
        activityQuantity = null;
        //order = null; // TODO
        spreaderData = null;
        profile = null;
        tour = null;
        streetstateNum = null;
        weatherNum = null;
        equipmentsSet = false;
        merits = new Merit[0];
        codriver = null;
    }


    private void setMissionEmployeeIfNotSet( int timeoutMS ) {
        try {
            if (mission!=null) {
                if (mission.getEmployee()==null) {
                    mission.setEmployee(
                            DbHelper.run(new Callable<Employee>() {
                                public Employee call() {
                                    return Session.getEmployee();
                                }
                            }).get(timeoutMS, TimeUnit.MILLISECONDS)
                    );
                }
            }
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            Log.e(TAG, "timout when trying to set employee" );
        }
    }

    public synchronized void setActivity(Activity activity) {
        if (missionActive.get()) {
            boolean activityChanged = (activity == null ? (this.activity != null) : !activity.equals(this.activity));
            if (activityChanged) {
                this.activity = activity;
                notify();
            }
        }
    }

    public synchronized void setActivityQuantity(Double quantity) {
        this.activityQuantity = quantity;
        notify();
    }

/*  TODO?
    public synchronized void setActivity(Order order, Activity activity) {
        if (missionActive.get()) {
            boolean orderChanged = (order == null ? this.order != null : !order.equals(this.order));
            boolean activityChanged = (activity == null ? (this.activity != null) : !activity.equals(this.activity));
            if (orderChanged || activityChanged) {
                this.order = order;
                this.activity = activity;
                notify();
            }
        }
    }*/

    public synchronized void setActivityNotice(String imagePath, String comment) {
        if (new EqualsBuilder().append(imagePath,activityImage).append(comment,activityComment).build()){
            return;
        }
        this.activityImage = imagePath;
        this.activityComment = comment;
        notify();
    }


    private void createMissionData() {
        Date now = new Date();

        // create mission-data
        MissionData missionData = new MissionData();
        missionData.setMissionId(mission.getId());
        if (location != null && Math.abs(location.getTime() - (now.getTime() - timeOffset)) < MAX_POSITION_AGE) {
            GpsData gpsData = GpsData.fromAndroidLocation(location);
            missionData.setTimestamp(gpsData.getTimestamp());
            missionData.setPosition(gpsData.getPosition());
            missionData.setSpeed(gpsData.getSpeed());
            Integer course = gpsData.getCourse();
            if (course!=null) {
                missionData.setCourse(course);
            }
        } else {
            missionData.setTimestamp(new Date(now.getTime() - timeOffset));
            missionData.setPosition(null);
        }
        missionData.setActivityId(activity == null ? null : activity.getId());
        //missionData.setOrderId(order == null ? null : order.getId()); TODO?
        missionData.setComment(activityComment);
        activityComment = null;
        if (activityImage != null) {
            boolean success=true;
            Bitmap bitmap = BitmapFactory.decodeFile(activityImage);
            if (bitmap==null){
                Log.e(TAG, "Could not decode bitmap" );
                success=false;
            }else{
                float factor = ((float) photoTargetWidth) / bitmap.getWidth();
                bitmap = Bitmap.createScaledBitmap(bitmap, (int) (bitmap.getWidth() * factor), (int) (bitmap.getHeight() * factor), false);
                ByteArrayOutputStream out = new ByteArrayOutputStream();
                if (!bitmap.compress(Bitmap.CompressFormat.JPEG, 70, out)){
                    Log.e(TAG, "Could not compress bitmap" );
                    success=false;
                }else{
                    missionData.setPhoto(out.toByteArray());
                }
            }
            if (success){
                new File(activityImage).delete();
            }
            activityImage = null;
        }
        if (activityQuantity != null) {
            missionData.setQuantity(activityQuantity);
            activityQuantity = null;
        }

        // set weather-data
        missionData.setWeatherNum(weatherNum);
        missionData.setStreetStateNum(streetstateNum);


        daoMissionData.insert(missionData);

        // set inputs for spreaderdata
        if (spreaderData!=null) {
            addInput(missionData, DigitalNumber.SPREADER, spreaderData.solidOn ? 1 : 0);
            addInput(missionData, AnalogNumber.SPREADER_WIDTH, spreaderData.solidWidth/100);
            addInput(missionData, AnalogNumber.SPREADER_DENSITY, spreaderData.solidDens);
            addInput(missionData, AnalogNumber.SPREADER_SOLE, spreaderData.solidBrine);
            addInput(missionData, DigitalNumber.SPRAYER, spreaderData.liquidOn ? 1 : 0);
            addInput(missionData, AnalogNumber.SPRAYER_WIDTH, spreaderData.liquidWidth/100);
            addInput(missionData, AnalogNumber.SPRAYER_DENSITY, spreaderData.liquidDens);
            if (spreaderData.temperature!=null)
                addInput(missionData, AnalogNumber.TEMPERATURE, spreaderData.temperature);
        }
        if (inputData!=null) {
            for (Map.Entry<Integer, Boolean> digiIn : inputData.digital.entrySet()) {
                for (DigitalNumber digiNumber : DigitalNumber.values()) {
                    if (digiNumber.getValue()==digiIn.getKey()){
                        addInput(missionData, digiNumber, digiIn.getValue()?1:0);
                        break;
                    }
                }
            }
            for (Map.Entry<Integer, Float> analIn : inputData.analog.entrySet()) {
                for (AnalogNumber analNumber : AnalogNumber.values()) {
                    if (analNumber.getValue()==analIn.getKey()){
                        addInput(missionData, analNumber, analIn.getValue());
                        break;
                    }
                }
            }
        }

        // set merits
        for (Merit merit : merits) {
            MissionDataMerit mdm = new MissionDataMerit();
            mdm.setMissionData(missionData);
            mdm.setMerit(merit);
            DaoMissionDataMerit.getInstance().insert(mdm);
        }

        // ok done.
        logger.debug("create missiondata");
        setMissionEmployeeIfNotSet((int) (MAX_POSITION_AGE/2));
    }

    private void addInput(MissionData missionData, InputNumber inputNumber, double value) {
        MissionDataInput input = new MissionDataInput();
        input.setAnalog(inputNumber instanceof AnalogNumber);
        input.setNumber(inputNumber.getValue());
        input.setValue(value);
        input.setMissiondata(missionData);
        DaoMissionDataInput.getInstance().insert(input);
    }

    @Override
    public synchronized void run() {
        try {
            createMissionData();
            while (missionActive.get()) {
                wait();
                createMissionData();
                locationFilter.updateLastLocation(location);
            }
        } catch (InterruptedException ex) {
            logger.error(ex);
        }
    }

    @Override
    public synchronized void onLocationChanged(@Nullable Location location) {
        if (location==null){
            // uh oh, we have problems with GPS-location.
            // TODO: maybe inform the user about it?
            return;
        }
        timeOffset = new Date().getTime() - location.getTime();
        if (Math.abs(timeOffset) > MAX_TIME_DRIFT) {
            Log.w(TAG, "time drift: " + timeOffset);
        }
        if (this.location!=null && location.getTime()<this.location.getTime()){
            Log.e(TAG, "locations poorly ordered: new location is older than previous location by "+(this.location.getTime()-location.getTime()));
            return;
        }
        this.location = location;
        if (locationFilter.filter(location) != null) {
            synchronized (this) {
                notify();
            }
        }
    }

    @Override
    public void onStatusChanged(GPSInfo gpsInfo) {}

    public boolean isMissionActive() {
        return missionActive.get();
    }


    private synchronized void notifyIfDiffers(Object newObj, Object oldObj) {
        if (differs(newObj,oldObj)){
            notify();
        }
    }
    private synchronized void notifyIfDiffers(Object[] newObjs, Object[] oldObjs) {
        if (differs(newObjs,oldObjs)){
            notify();
        }
    }
    private synchronized boolean differs(Object newObj, Object oldObj) {
        if (oldObj == null){
            if (newObj != null){
                return true;
            }
        }else{
            if (newObj == null){
                return true;
            }else{
                if (!oldObj.equals(newObj)){
                    return true;
                }
            }
        }
        return false;
    }
    private synchronized boolean differs(Object[] newObjs, Object[] oldObjs) {
        equalsBuilder.reset();
        return equalsBuilder.append(newObjs,oldObjs).isEquals();
    }

    @Override
    public synchronized void onSpreaderData(SpreaderData newSpreaderData) {
        SpreaderData oldSpreaderData = spreaderData;
        spreaderData=newSpreaderData;
        notifyIfDiffers(newSpreaderData,oldSpreaderData);
    }

    @Override
    public synchronized void onInputData(@Nullable InputData newInputData) {
        InputData oldInputData = inputData;
        inputData=newInputData;
        if (differs(newInputData,oldInputData)){
            if (newInputData==null || oldInputData==null){
                notify();
            }else{
                if (!newInputData.equalsDelta(oldInputData)){
                    notify();
                }
            }
        }
    }

    @Override
    public void onMobidatStatus(boolean spreader, boolean inputs, boolean sending) {}

    public synchronized void setProfile(Profile profile) {
        this.profile = profile;
        if (mission!=null){
            mission.setProfile(profile);
            daoMission.update(mission);
        }
    }

    public synchronized void setTour(Tour tour){
        this.tour = tour;
        if (mission!=null){
            mission.setTour(tour);
            daoMission.update(mission);
        }
    }

    public synchronized void setWeatherNum(Integer newWeatherNum) {
        Integer oldWeatherNum = weatherNum;
        weatherNum = newWeatherNum;
        notifyIfDiffers(newWeatherNum, oldWeatherNum);
    }

    public synchronized void setStreetstateNum(Integer newStreetstateNum) {
        Integer oldStreetstateNum = streetstateNum;
        streetstateNum = newStreetstateNum;
        notifyIfDiffers(newStreetstateNum, oldStreetstateNum);
    }

    public synchronized void setMerits(Merit[] newMerits) {
        Merit[] oldMerits = merits;
        merits = newMerits;
        notifyIfDiffers(newMerits,oldMerits);
    }
    public synchronized void setCodriver(Employee newCodriver) {
        this.codriver = codriver;
        if (mission!=null){
            mission.setCodriver(newCodriver);
            daoMission.update(mission);
        }
    }

    public synchronized void setEquipments() {
        this.equipmentsSet = true;
        if (mission!=null){
            DaoMissionEquipment daoMissionEquipment = DaoMissionEquipment.getInstance();
            daoMissionEquipment.deleteForMission(mission);
            for (MissionEquipment me : daoMissionEquipment.getLastUsed()) {
                MissionEquipment ourMe = new MissionEquipment(me);
                ourMe.setMission(mission);
                daoMissionEquipment.saveOrUpdate(ourMe);
            }
        }
    }

}