package org.elasticflow.connection.sockets;

import java.net.URI;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.elasticflow.config.GlobalParam.END_TYPE;
import org.elasticflow.connection.EFConnectionSocket;
import org.elasticflow.param.pipe.ConnectParams;
import org.elasticflow.param.warehouse.WarehouseParam;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Hdfs basic connection establishment management class
 * @author chengwen
 * @version 1.0
 * @date 2021-06-24 09:25
 */

public class HdfsConnection extends EFConnectionSocket<FileSystem> {
	
	final String DEFAULT_KEY = "username";
	
	private final static Logger log = LoggerFactory.getLogger("Hdfs Socket");
	
	public static EFConnectionSocket<?> getInstance(ConnectParams ConnectParams) {
		EFConnectionSocket<?> o = new HdfsConnection();
		o.init(ConnectParams);
		return o;
	}
	
	@Override
	public boolean connect(END_TYPE endType) {
		WarehouseParam wnp = this.connectParams.getWhp();
		if (wnp.getHost() != null) {
			if (!status()) { 			        	
				Configuration conf = new Configuration();
				conf.set("fs.defaultFS", wnp.getHost());
		        try {
					this.conn = FileSystem.get(new URI(wnp.getHost()), conf, 
							wnp.getDefaultValue().getString(DEFAULT_KEY));
				} catch (Exception e) {
					log.error("{} get hdfs {} connect exception,", wnp.getAlias(),endType.name(),e);
					this.conn = null;
				}
			}
		} else {
			return false;
		}
		return true;
	}


	@Override
	public boolean status() {
		if(this.conn==null)
			return false;
		return true;
	}

	@Override
	public boolean free() {
		try {
			if(this.conn!=null)
				this.conn.close();
			this.conn = null;
			this.connectParams = null;
		} catch (Exception e) {
			log.warn("{} free hdfs connection exception", this.connectParams.getWhp().getAlias(),e);
			return false;
		}
		return true;
	}

}
