package fr.d2si.ooso.mappers_listener;

import com.amazonaws.services.lambda.runtime.Context;
import com.amazonaws.services.lambda.runtime.RequestHandler;
import fr.d2si.ooso.reducers_driver.ReducersDriverInfo;
import fr.d2si.ooso.utils.Commons;
import fr.d2si.ooso.utils.JobInfo;
import fr.d2si.ooso.utils.JobInfoProvider;

public class MappersListener implements RequestHandler<Void, String> {
    private static final int HEARTBEAT_INTERVAL = 500;

    @Override
    public String handleRequest(Void aVoid, Context context) {
        try {
            JobInfo jobInfo = JobInfoProvider.getJobInfo();

            int currentMappersOutputFiles = Commons.getBucketObjectSummaries(jobInfo.getMapperOutputBucket(), jobInfo.getJobId()).size();
            int expectedMappersOutputFiles = Commons.getBucketObjectSummaries(jobInfo.getJobInputBucket()).size();

            if (!jobInfo.getDisableReducer()) {
                if (currentMappersOutputFiles == expectedMappersOutputFiles)
                    invokeReducerCoordinator();
                else {
                    Thread.sleep(HEARTBEAT_INTERVAL);
                    invokeMappersListener();
                }
            }

            return String.valueOf(currentMappersOutputFiles == expectedMappersOutputFiles);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    private void invokeMappersListener() {
        Commons.invokeLambdaAsync("mappers_listener", null);
    }

    private void invokeReducerCoordinator() {
        ReducersDriverInfo reducersDriverInfo = new ReducersDriverInfo(0);
        Commons.invokeLambdaAsync("reducers_driver", reducersDriverInfo);
    }

}

@@ -632,32 +632,83 @@
 ersOutputFiles =
+%0A                    !jobInfo.getDisableReducer() ?
  Commons.getBuck
@@ -784,17 +784,142 @@
 ).size()
-;
+ :%0A                            Commons.getBucketObjectSummaries(jobInfo.getReducerOutputBucket(), jobInfo.getJobId()).size();%0A
 %0A       
@@ -1031,60 +1031,8 @@
 );%0A%0A
-            if (!jobInfo.getDisableReducer()) %7B%0A    
     
@@ -1099,16 +1099,68 @@
 utFiles)
+ %7B%0A                if (!jobInfo.getDisableReducer())
 %0A       
@@ -1216,19 +1216,17 @@
     
-   
+%7D
  else %7B%0A
     
@@ -1221,20 +1221,16 @@
  else %7B%0A
-    
         
@@ -1283,28 +1283,24 @@
             
-   
- 
 invokeMapper
@@ -1316,26 +1316,8 @@
 ();%0A
-                %7D%0A
     

Fix mappers listener