package org.springblade.sink;

import com.mongodb.MongoClient;
import com.mongodb.MongoCredential;
import com.mongodb.ServerAddress;
import com.mongodb.client.MongoDatabase;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.bson.Document;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;

/**
 * @author: lianghanqiang
 * @description: 自定义 sink for mongodb
 * @since: 10/26/21 -- 11:12 AM
 */
public class MongoSink extends RichSinkFunction<String> {

	private MongoClient mongoClient;
	private MongoDatabase mongoDatabase;

	/**
	 * 链接mongo
	 */
	@Override
	public void open(Configuration parameters) throws Exception {
		if(Objects.isNull(mongoDatabase)){
			//通过连接认证获取MongoDB连接
			ServerAddress serverAddress = new ServerAddress("192.168.1.218",27017);
			List<ServerAddress> addrs = new ArrayList<ServerAddress>();
			addrs.add(serverAddress);

			//MongoCredential.createScramSha1Credential()三个参数分别为 用户名 数据库名称 密码
			MongoCredential credential = MongoCredential.createScramSha1Credential("root", "ldt", "123456".toCharArray());
			List<MongoCredential> credentials = new ArrayList<MongoCredential>();
			credentials.add(credential);

			//通过连接认证获取MongoDB连接
			this.mongoClient = new MongoClient(addrs,credentials);
			
		}
		super.open(parameters);
	}

	/**
	 * 将数据写入
	 */
	@Override
	public void invoke(String value, Context context) throws Exception {
		System.out.println(value);
		Document document = new Document();
		document.append("data","data");
		this.mongoDatabase.getCollection("log").insertOne(document);
	}

	/**
	 * 释放资源
	 */
	@Override
	public void close() throws Exception {
		this.mongoClient.close();
		super.close();
	}
}
