/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements. See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership. The ASF 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.javierdallamore.camel.component.eventfabric;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.nio.ByteBuffer;
import java.nio.charset.Charset;

import org.apache.camel.Exchange;
import org.apache.camel.impl.DefaultProducer;
import org.codehaus.jackson.map.ObjectMapper;
import org.codehaus.jackson.node.ObjectNode;

import com.eventfabric.api.client.EndPointInfo;
import com.eventfabric.api.client.EventClient;
import com.eventfabric.api.client.Response;
import com.eventfabric.api.model.Event;
import com.sun.xml.fastinfoset.Encoder;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @version $Revision: 1.1 $
 */
public class EventFabricProducer extends DefaultProducer {
	private static final Logger LOG = LoggerFactory
			.getLogger(EventFabricProducer.class);
	private final EventFabricEndpoint endpoint;
	private ObjectMapper mapper = new ObjectMapper();
	private int attemps = 0;
	
	public EventFabricProducer(EventFabricEndpoint endpoint) {
		super(endpoint);
		this.endpoint = endpoint;
	}

	/**
	 *
	 * @param exchange
	 * @throws Exception
	 */
	@Override
	public void process(Exchange exchange) throws Exception {
		attemps += 1;
		byte[] bs = exchange.getIn().getBody(byte[].class);
		String body = "{\"data\": []}";
		try {
			body = String.format("{\"data\": %s}", new String(bs, "UTF-8"));
		} catch(UnsupportedEncodingException uee) {
		    LOG.error(uee.getMessage());
		}
		ObjectNode value = (ObjectNode) mapper.readTree(body);
		String channel = endpoint.getChannel();
		EventClient eventClient = endpoint.getEventClient();
		if (channel == null) {
			channel = endpoint.getName();
		}
		try {
			Event event = new Event(channel, value);
			Response response = eventClient.send(event);
			if (response.getStatus() == 201) {
				LOG.info(String.format("%s sent to Event Fabric", endpoint.getName()));
			} else if (response.getStatus() == 401 && attemps <= 3) {
				LOG.error(String.format("Event Fabric session expired. Trying to log in again. Attemp: %d", attemps));
				eventClient.authenticate();
				process(exchange);
			} else {
				LOG.error(String.format("Error sending %s to Event Fabric: %s", endpoint.getName(), response.getResult()));
			}
		} catch (IOException e) {
			LOG.error(e.getMessage());
		} finally {
			attemps = 0;
		}
	}
}

@@ -950,24 +950,52 @@
 ByteBuffer;%0A
+import java.nio.CharBuffer;%0A
 import java.
@@ -2036,17 +2036,33 @@
 ;%0A%09%09
+%0A%09%09%0A%09%09/*%0A%09%09 
 byte%5B%5D 
-bs
+latin1
  = e
@@ -2106,200 +2106,491 @@
 ;%0A%09%09
-String body = %22%7B%5C%22data%5C%22: %5B%5D%7D%22;%0A%09%09try %7B%0A%09%09%09body = String.format(%22%7B%5C%22data%5C%22: %25s%7D%22, new String(bs, %22UTF-8%22));%0A%09%09%7D catch(UnsupportedEncodingException uee) %7B%0A%09%09    LOG.error(uee.getMessage()
+ Charset utf8charset = Charset.forName(%22UTF-8%22);%0A%09%09Charset iso88591charset = Charset.forName(%22ISO-8859-1%22);%0A%09%09ByteBuffer inputBuffer = ByteBuffer.wrap(latin1);%0A%09%09// decode UTF-8%0A%09%09CharBuffer data = iso88591charset.decode(inputBuffer);%0A%09%09// encode ISO-8559-1%0A%09%09ByteBuffer outputBuffer = utf8charset.encode(data);%0A%09%09byte%5B%5D outputData = outputBuffer.array();%0A%09%09%0A%09%09%0A%09%09*/%0A%09%09%0A%09%09String body = exchange.getIn().getBody(String.class);%0A%09%09String data = String.format(%22%7B%5C%22data%5C%22: %25s%7D%22, body
 );%0A%09%09
-%7D
 %0A%09%09O
@@ -2636,20 +2636,20 @@
 eadTree(
-body
+data
 );%0A%09%09Str

read body as string in camel eventfabric