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 (currentMappersOutputFiles == expectedMappersOutputFiles) {
                if (!jobInfo.getDisableReducer())
                    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);
    }

}

@@ -847,24 +847,76 @@
 )).size();%0A%0A
+            if (!jobInfo.getDisableReducer()) %7B%0A    
             
@@ -979,60 +979,8 @@
 les)
- %7B%0A                if (!jobInfo.getDisableReducer())
 %0A   
@@ -1036,17 +1036,19 @@
         
-%7D
+   
  else %7B%0A
@@ -1063,16 +1063,20 @@
         
+  
+  
 Thread.s
@@ -1113,24 +1113,28 @@
             
+    
 invokeMapper
@@ -1142,24 +1142,42 @@
 Listener();%0A
+                %7D%0A
             

Fix reversed condition