/*
 * Copyright (c) 2017, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
 *
 * WSO2 Inc. licenses this file to you 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 org.wso2.carbon.transport.file.test.util;

import org.wso2.carbon.messaging.BinaryCarbonMessage;
import org.wso2.carbon.messaging.CarbonCallback;
import org.wso2.carbon.messaging.CarbonMessage;
import org.wso2.carbon.messaging.CarbonMessageProcessor;
import org.wso2.carbon.messaging.ClientConnector;
import org.wso2.carbon.messaging.TransportSender;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * Message processor that is used for the testing purposes.
 */
public class FileMessageProcessor implements CarbonMessageProcessor {
    private CountDownLatch latch = new CountDownLatch(1);
    private ExecutorService executor = Executors.newSingleThreadExecutor();
    private List<String> fileContent = new ArrayList<>();

    @Override
    public boolean receive(CarbonMessage carbonMessage, CarbonCallback carbonCallback) throws Exception {
        executor.execute(new Runnable() {
            @Override public void run() {
                byte[] content = ((BinaryCarbonMessage) carbonMessage).readBytes().array();
                fileContent.add(new String(content));

            }
        });
        latch.await(10, TimeUnit.MICROSECONDS);
        if (fileContent.size() == 20) {
            done();
        }
        return false;
    }

    @Override
    public void setTransportSender(TransportSender transportSender) {

    }

    @Override
    public void setClientConnector(ClientConnector clientConnector) {

    }

    @Override
    public String getId() {
        return "test-file-message-processor";
    }

    public String getFileContent(int index) {
        return fileContent.get(index);
    }

    /**
     * To wait till file reading operation is finished.
     *
     * @throws InterruptedException Interrupted Exception.
     */
    public void waitTillDone() throws InterruptedException {
        latch.await();
    }

    /**
     * To make sure the reading the file content is done.
     */
    private void done() {
        latch.countDown();
    }

}

@@ -1709,302 +1709,143 @@
     
-executor.execute(new Runnable() %7B%0A            @Override public void run() %7B%0A                byte%5B%5D content = ((BinaryCarbonMessage) carbonMessage).readBytes().array();%0A                fileContent.add(new String(content));%0A%0A            %7D%0A        %7D);%0A        latch.await(10, TimeUnit.MICROSECONDS
+CountDownLatch latch = new CountDownLatch(1);%0A        executor.execute(new FillFileContent(latch, carbonMessage));%0A        latch.await(
 );%0A 
@@ -2676,10 +2676,540 @@
 %0A    %7D%0A%0A
+    private class FillFileContent implements Runnable %7B%0A        CountDownLatch latch;%0A        CarbonMessage carbonMessage;%0A%0A        public FillFileContent(CountDownLatch latch, CarbonMessage carbonMessage) %7B%0A            this.latch = latch;%0A            this.carbonMessage = carbonMessage;%0A        %7D%0A%0A        @Override%0A        public void run() %7B%0A            byte%5B%5D content = ((BinaryCarbonMessage) carbonMessage).readBytes().array();%0A            fileContent.add(new String(content));%0A            latch.countDown();%0A        %7D%0A    %7D%0A
 %7D%0A

Fix file server connecter test hanging issue