package org.springblade.binlog.listener;

import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.Event;
import com.github.shyiko.mysql.binlog.event.EventData;
import com.github.shyiko.mysql.binlog.event.QueryEventData;
import com.github.shyiko.mysql.binlog.event.TableMapEventData;
import com.github.shyiko.mysql.binlog.event.WriteRowsEventData;
import com.github.shyiko.mysql.binlog.event.deserialization.EventDeserializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.io.Serializable;
import java.util.List;
public class BinlogListenerMixed {

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

	private static final String MYSQL_HOST = "127.0.0.1";
	private static final int MYSQL_PORT = 3308;
	private static final String MYSQL_USERNAME = "root";
	private static final String MYSQL_PASSWORD = "root";

	public static void main(String[] args) {
		try {
			BinaryLogClient client = new BinaryLogClient(MYSQL_HOST, MYSQL_PORT, MYSQL_USERNAME, MYSQL_PASSWORD);
//            client.setBinlogFilename(null);
//            client.setBinlogPosition(-1); // 或者设置为其他适当的初始位置
//            client.setServerId(1);
//            client.setBinlogFilename("mysql-bin.000005");
//            client.setBinlogPosition(154);
			EventDeserializer eventDeserializer = new EventDeserializer();
			eventDeserializer.setCompatibilityMode(
				EventDeserializer.CompatibilityMode.DATE_AND_TIME_AS_LONG
				// EventDeserializer.CompatibilityMode.CHAR_AND_BINARY_AS_BYTE_ARRAY 该设置会将varchar 转为 byte[]
			);
			logger.info("使用主机={}, 端口={}, 用户名={}, 密码={} 连接到 MySQL", MYSQL_HOST, MYSQL_PORT, MYSQL_USERNAME, MYSQL_PASSWORD);
			client.setEventDeserializer(eventDeserializer);
			client.registerEventListener(BinlogListenerMixed::handleEvent);
			client.registerLifecycleListener(new BinaryLogClient.LifecycleListener() {
				@Override
				public void onConnect(BinaryLogClient client) {
					logger.info("Connected to MySQL server");
				}

				@Override
				public void onCommunicationFailure(BinaryLogClient client, Exception ex) {
					logger.error("Communication failure with MySQL server", ex);
				}

				@Override
				public void onEventDeserializationFailure(BinaryLogClient client, Exception ex) {
					logger.error("Event deserialization failure", ex);
				}

				@Override
				public void onDisconnect(BinaryLogClient client) {
					logger.warn("Disconnected from MySQL server");
					// 在这里添加重新连接或其他处理逻辑
				}
			});

			client.connect();
		} catch (IOException e) {
			logger.error("@@ 连接到 MySQL 时发生错误", e);
			logger.error("@@ Error connecting to MySQL", e);
		}
	}

	private static void handleEvent(Event event) {
		logger.info("@@ 打印 event: {}", event);
		logger.info("@@ Received event type: {}", event.getHeader().getEventType());

		switch (event.getHeader().getEventType()) {
			case WRITE_ROWS:
			case EXT_WRITE_ROWS:
				handleWriteRowsEvent((WriteRowsEventData) event.getData());
				break;
			case QUERY:
				handleQueryEvent((QueryEventData) event.getData());
				break;
			case TABLE_MAP:
				handleTableMapEvent((TableMapEventData) event.getData());
				break;
			// 其他事件处理...
		}
	}

	private static void handleWriteRowsEvent(WriteRowsEventData eventData) {
		List<Serializable[]> rows = eventData.getRows();

		// 获取表名
		String tableName = getTableName(eventData);

		// 处理每一行数据
		for (Serializable[] row : rows) {
			// 根据需要调整以下代码以获取具体的列值
			String column1Value = row[0].toString();
			String column2Value = row[1].toString();
			String url = row[2].toString();
			logger.info(url);

			// 将数据备份到另一个数据库
			backupToAnotherDatabase(tableName, column1Value, column2Value);
		}
	}

	private static void handleQueryEvent(QueryEventData eventData) {
		String sql = eventData.getSql();
		logger.info("@@ handleQueryEvent函数执行Query event SQL: {}", sql);

		// 解析SQL语句，根据需要处理
		// 例如，检查是否包含写入操作，然后执行相应的逻辑
	}

	private static void handleTableMapEvent(TableMapEventData eventData) {
		// 获取表映射信息，根据需要处理
		logger.info("@@ handleTableMapEvent函数执行TableMap event: {}", eventData);
	}

	private static String getTableName(EventData eventData) {
		// 获取表名的逻辑，可以使用TableMapEventData等信息
		// 根据实际情况实现
		return "example_table";
	}

	private static void backupToAnotherDatabase(String tableName, String column1Value, String column2Value) {
		// 将数据备份到另一个数据库的逻辑
		logger.info("Backup to another database: Table={}, Column1={}, Column2={}", tableName, column1Value, column2Value);
	}
}