/**
 * Copyright (C) 2014 Microsoft Corporation
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *         http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.microsoft.reef.io.data.loading.api;

import com.microsoft.reef.driver.evaluator.EvaluatorRequest;
import com.microsoft.reef.driver.evaluator.EvaluatorRequestor;
import com.microsoft.wake.EventHandler;

import java.util.concurrent.CountDownLatch;
import java.util.logging.Level;
import java.util.logging.Logger;

public class ResourceRequestHandler implements EventHandler<EvaluatorRequest> {

  private static final Logger LOG = Logger.getLogger(ResourceRequestHandler.class.getName());

  private final EvaluatorRequestor requestor;

  private CountDownLatch resourceRequestGate = new CountDownLatch(1);

  public ResourceRequestHandler(final EvaluatorRequestor requestor) {
    this.requestor = requestor;
  }

  public void releaseResourceRequestGate() {
    LOG.log(Level.FINE, "Releasing Gate");
    this.resourceRequestGate.countDown();
  }

  @Override
  public void onNext(final EvaluatorRequest request) {

    LOG.log(Level.FINE, "Processing a request with count: {0} - Waiting for gate to be released",
        request.getNumber());

    for (;;) {
      try {
        this.resourceRequestGate.await();
        break;
      } catch (final InterruptedException ex) {
        LOG.log(Level.FINEST, "Interrupted", ex);
      }
    }

    LOG.log(Level.FINE, "Gate released. Submitting request: {0}", request);
    this.resourceRequestGate = new CountDownLatch(1);
    this.requestor.submit(request);
  }
}

@@ -1534,17 +1534,29 @@
 uest) %7B%0A
-%0A
+    try %7B%0A%0A  
     LOG.
@@ -1570,16 +1570,26 @@
 el.FINE,
+%0A         
  %22Proces
@@ -1663,16 +1663,18 @@
         
+  
 request.
@@ -1692,37 +1692,8 @@
 );%0A%0A
-    for (;;) %7B%0A      try %7B%0A  
     
@@ -1732,136 +1732,11 @@
 ();%0A
+%0A
   
-      break;%0A      %7D catch (final InterruptedException ex) %7B%0A        LOG.log(Level.FINEST, %22Interrupted%22, ex);%0A      %7D%0A    %7D%0A%0A
     
@@ -1807,24 +1807,26 @@
 quest);%0A    
+  
 this.resourc
@@ -1863,24 +1863,26 @@
 tch(1);%0A    
+  
 this.request
@@ -1900,15 +1900,116 @@
 equest);
+%0A%0A    %7D catch (final InterruptedException ex) %7B%0A      LOG.log(Level.FINEST, %22Interrupted%22, ex);%0A    %7D
 %0A  %7D%0A%7D%0A

remove the infinite loop around .await() call