/*
 * Copyright (c) 2016, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
 *
 * 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 org.wso2.carbon.gateway.core.flow.contentaware;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.wso2.carbon.gateway.core.config.ConfigRegistry;
import org.wso2.carbon.gateway.core.flow.contentaware.abstractcontext.TypeConverter;
import org.wso2.carbon.gateway.core.flow.contentaware.exceptions.TypeConversionException;
import org.wso2.carbon.messaging.CarbonMessage;

import java.io.IOException;
import java.io.InputStream;
import java.nio.ByteBuffer;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * A class to decouple the mediators from the TypeConverterRegistry and to handle
 * the conversions.
 */
public class ConversionManager {
    private static final Logger log = LoggerFactory.getLogger(ConversionManager.class);

    private static ConversionManager instance = new ConversionManager();

    public static ConversionManager getInstance() {
        return instance;
    }

    private ConversionManager() {};

    public InputStream convertTo(CarbonMessage cMsg, String sourceType, String targetType) {

        TypeConverter converter = ConfigRegistry.getInstance()
                .getTypeConverterRegistry().lookup(targetType, sourceType);

        if (converter == null) {
            if (log.isDebugEnabled()) {
                log.debug("No type converted found for Source: " + sourceType + " Target : " + targetType);
            }
            return null;
        }

        //Aggregation and creating inputStream
        BlockingQueue<ByteBuffer> contentBuf = aggregateContent(cMsg);
        InputStream inputStream = new ByteBufferBackedInputStream(contentBuf);
        InputStream processedStream = null;

        try {
            processedStream = converter.convert(inputStream);
        } catch (TypeConversionException e) {
            log.error("Error in converting from: " + sourceType + " to: " + targetType);
        } catch (IOException e) {
            log.error("Error " + e);
        } finally {
            //TODO: do we need to close the input stream here ?
        }
        return processedStream;
    }

    private BlockingQueue<ByteBuffer> aggregateContent(CarbonMessage msg) {

        try {
            //Check whether the message is fully read
            while (!msg.isEndOfMsgAdded()) {
                Thread.sleep(10);               // TODO: why thread sleeps ?
            }
            //Get a clone of content chunk queue from the pipe
            BlockingQueue<ByteBuffer> clonedContent = new LinkedBlockingQueue<>(msg.getFullMessageBody());
            return clonedContent;
        } catch (Exception e) {
            log.error("Error while cloning ", e);
        }
        return null;
    }
}

@@ -1111,24 +1111,50 @@
 ByteBuffer;%0A
+import java.util.Scanner;%0A
 import java.
@@ -1657,17 +1657,28 @@
 ager() %7B
-%7D
+%0A    %7D%0A%0A    
 ;%0A%0A    p
@@ -1683,27 +1683,29 @@
  public 
-InputStream
+CarbonMessage
  convert
@@ -1738,40 +1738,129 @@
 ing 
-sourceType, String targetType) %7B
+targetType) %7B%0A%0A        String sourceType = cMsg.getHeader(%22Content-Type%22);%0A        StringBuffer buf = new StringBuffer();
 %0A%0A  
@@ -1919,16 +1919,43 @@
 stance()
+.getTypeConverterRegistry()
 %0A       
@@ -1984,25 +1984,8 @@
 rter
-Registry().lookup
 (tar
@@ -2868,228 +2868,376 @@
     
-return processedStream
+Scanner s = new java.util.Scanner(processedStream).useDelimiter(%22%5C%5CA%22)
 ;%0A    
-%7D%0A%0A
     
-private BlockingQueue%3CByteBuffer%3E aggregateContent(CarbonMessage msg) %7B%0A%0A        try %7B%0A            //Check whether the message is fully read%0A            while (!msg.isEndOfMsgAdded()) %7B
+while (s.hasNext()) %7B%0A            buf.append(s.next());%0A        %7D%0A        String outputString = buf.toString();%0A        log.info(sourceType + %22 to %22 + targetType + %22 type conversion%5Cn%22 + outputString);%0A        ByteBuffer outputByteBuffer = ByteBuffer.wrap(String.valueOf(buf).getBytes());
 %0A   
@@ -3245,41 +3245,107 @@
     
-        Thread.sleep(10
+cMsg.setHeader(%22Content-Type%22, targetType);%0A        cMsg.addMessageBody(outputByteBuffer
 );
+%0A
         
     
@@ -3344,43 +3344,105 @@
     
-       // TODO: why thread sleeps ?
+return cMsg;%0A%0A    %7D%0A%0A    private BlockingQueue%3CByteBuffer%3E aggregateContent(CarbonMessage msg) %7B%0A
 %0A   
@@ -3438,37 +3438,37 @@
 msg) %7B%0A%0A        
-    %7D
+try %7B
 %0A            //G

add mediator work to conversion manager