data-server-java/src/main/java/com/tidecloud/dataacceptance/service/impl/BingShuiGpsServerHandler.java

package com.tidecloud.dataacceptance.service.impl;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;

import com.tidecloud.dataacceptance.service.HexBinaryAcceptanceHandlerAdapter;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandler;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;

/**
 * @author cdk
 */
@Component
@Scope("prototype")
@ChannelHandler.Sharable
public class BingShuiGpsServerHandler extends HexBinaryAcceptanceHandlerAdapter {

	private static final Logger logger = LoggerFactory.getLogger(BingShuiGpsServerHandler.class);

	public void handle(String oringalData, Channel channel) {
		
	}

	@Override
	protected void handle(ByteBuf in, Channel channel) throws Exception {
		ByteBuf byteBuf = (ByteBuf) in;
		String oringalData = byteBufferToString(byteBuf.nioBuffer());
		logger.info("接入数据：{}", oringalData);
		Map<String, String> dataMap = new HashMap<>();

		String[] dataArray = oringalData.split("\\s+");
		String phone = dataArray[0];
		String deviceData = dataArray[1];
		if (deviceData.endsWith("]") && deviceData.startsWith("[")) {
			String subDeviceData = deviceData.substring(1, deviceData.length() - 1);
			String[] subDataArray = subDeviceData.split(";");
			try {
				for (int i = 0; i < subDataArray.length; i++) {
					String subData = subDataArray[i];
					String[] datas = subData.split(":");
					if (datas[0].contains("S") && !datas[0].contains("G")) {
						dataMap.put("S", datas[1]);
					} else {
						dataMap.put(datas[0], datas[1]);
					}
				}
			} catch (Exception e) {
				logger.error(e.getMessage(), e);
			}
		}
		
		byte[] dataByteArray = new byte[in.readableBytes()];
		in.readBytes(dataByteArray);
		sendMsg2Kafka(dataByteArray,phone,channel);


		String serialStr = dataMap.get("S");
		Integer serialId = Integer.valueOf(serialStr) + 1;
		Date now = new Date();
		SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyyMMddHHmmss");
		String dataStr = simpleDateFormat.format(now);
		String replyMsg = "[" + "S:" + serialId + ";R" + "1" + ";" + "ST:" + dataStr + "]";
		ByteBuf buffer = Unpooled.buffer(replyMsg.getBytes().length);
		buffer.writeBytes(replyMsg.getBytes());

		ChannelFuture channelFuture = channel.writeAndFlush(buffer);
		channelFuture.addListener(new GenericFutureListener<Future<? super Void>>() {
			@Override
			public void operationComplete(Future<? super Void> future) throws Exception {
				logger.info("回复设备: [{}] 成功", replyMsg);
			}
		});
	}

}