// UsingPattern,要求对象必须有close方法并且必定调用其close方法
usingPattern(HDFS){ hdfs => 
    hdfs.exists(new Path(config.path))
}


// 本地文件系统IO
BootsDefault.fs
	.config.path(path).build()
	.read().length == fileDataList.length
	

// Http Utils	
BootsDefault.http.get("https://www.baidu.com/")	

// format
BootsDefault.format.byteSize(1232786734)
BootsDefault.format.tryDate(new Date()) // if null default '-'

// convert
// 默认使用配置 nel.env.date.in.formats 转换 
BootsDefault.convert.date("2018-1-1")    

// 对象池  
val pool = new ObjectPool[BootsConnection](cfg, creator: () => T)
.....

# 配置
# test_db db映射(支持多数据库)
nel.orm.db.test_db.mysql.url=jdbc:mysql://127.0.0.1:3306/test_db_name?serverTimezone=UTC&characterEncoding=UTF-8&useSSL=false
nel.orm.db.test_db.mysql.user=root
nel.orm.db.test_db.mysql.password=12abAB

//对象映射
//使用MySQL注解完成db映射和方言映射  
@MySQL(db = "test_db",table = "t_test",isColumnAutoMapping = true)
class TestDomain extends SomeParentDomain{
  var a:String = null
  
  def set(a:String):this.type = {
      this.a = a
      this
  }
}

/**
* 命令系-CRUD 
*/
import nel.framework.boots.orm.BootsOrm._

// createNew 创建一个实体(补全主键生成,CreateBy,Creator等等)
// saveOrUpdate 新增或更新  
// find 根据主键查找一个实体  
// load 根据主键加载一个实体(约定如果查找为null将抛出异常)
// delete 根据主键删除一个实体
// ......
// commit 显式提交当前事务(目标方言需支持事务,不显式提交将在unitOfWork尾自动调用) 
// rollback 显式回滚当前事务(目标方言需支持回滚,unitOfWork出现异常将自动回滚)

val id = "id"
BootsDefault.orm.unitOfWork("Operator")(implicit sessionContext => {
  sessionContext.createNew[TestDomain](id).set("aaa") 
})
BootsDefault.orm.unitOfWork("Operator")(implicit sessionContext => {
  sessionContext.saveOrUpdate {
	sessionContext.load[TestDomain](id).set("bbb")
  }
})


/**
* 查询系-依赖目标方言
*/

BootsDefault.orm.unitOfWork("Operator")(implicit sessionContext => {

   import nel.framework.boots.orm.dialect.mysql.MySQLDialect._
   //指定目标方言,不同例如MySQL 和 HBase 的查询系API是不同

   // mysqlExecuteByClazzMeta 执行SQL
   // mysqlQueryListByDB 指定DB执行SQL查询列表
   // mysqlQuerySingleByClazzMeta 依赖ClazzMappingMeta执行SQL查询首行
   // mysqlQueryListByClazzMeta 依赖ClazzMappingMeta执行SQL查询列表
   // mysqlQueryConditionListByClazzMeta 依赖ClazzMappingMeta执行SQL条件查询列表
   // ....

  //依据条件Tuple查询列表
  sessionContext.mysqlQueryConditionListByClazzMeta[TestDomain](("a", "aaa"))

  //依赖ClazzMappingMeta执行SQL查询
  sessionContext.mysqlQueryListByClazzMeta[TestDomain](
	"select * from test_db where a =? ;",
	ListBuffer("aaa")
  )
  
  //执行SQL查询
  sessionContext.mysqlQueryListByDB(
	db = "test_db",
	sql = "select * from test_db where a =? ;",
	parameters = ListBuffer("aaa")
  )(rs=> new TestDomain(){.....})
})

// 为Boots加载Redis插件  
import nel.framework.boots.modules.redis.BootsRedis._

/**
* 写入
*/

BootsDefault.cache.redis
	.set("test_key","123") // 使用默认过期超时10秒  
	
BootsDefault.cache.redis
  .config.host("xxxx").port(1234).build()
  .set("test_key","123",30,TimeUnit.SECONDS)		
  
/**
* 读取
*/
BootsDefault.cache.redis.get[Test]("test_key")

//删除测试库
BootsDefault.influxDB.execute("drop measurement \"cpu\"")

//批量写入数据
BootsDefault.influxDB.batch(Array(
Point.measurement("cpu")
  .time(System.currentTimeMillis(), TimeUnit.MILLISECONDS)
  .addField("idle", 90L)
  .addField("user", 9L)
  .addField("system", 1L)
  .build()
))

//查询
BootsDefault.influxDB.execute("SELECT idle FROM cpu").size > 0

// 为Boots加载HDFS向插件
import cn.night.nel.framework.boots.modules.fs.BootsHDFS._

val path = "hdfs://XXX:9000/test/file_suit/fileSuitOutput.txt"

// 追加写入  
BootsDefault.hdfs
        .config.path(path).saveMod(SaveMode.Append).build()
        .write(fileDataList)
        
// 覆盖写入
BootsDefault.hdfs
        .config.path(path).saveMod(SaveMode.Overwrite).build()
        .write(fileDataList)

// 读取  
val list:List[String] = BootsDefault.hdfs
        .config.path(path).build()
        .read()

val path = "/test/node"

BootsDefault.zk.
	.config.build
	.createWithCreatingParents(CreateMode.EPHEMERAL,path,"123")

BootsDefault.zk.
	.config.build
	.setData(path,"456")
	
BootsDefault.zk.
	.config.build
	.getData(path)	

BootsDefault.zk.
	.config.build
	.getChildren("/test")		
	
BootsDefault.zk.
	.config.build
	.delete(path)

//发送
val sender = BootsDefault.kafka.sender
      .config.kafkaUrl("192.168.18.151:9092").build()
      .sendString(
          topic = "kafka.consumer.test",
          message = "M" + idx,
          key = Some("kafka.consumer.test"))
//接收          
BootsDefault.kafka.sequenceConsumer()
  .config
	.kafkaUrl("xxx:9092")
	.kafkaConsumerTopics("your_topic")
	.kafkaConsumerGroup("your_group")
  .build()
  .process(messages => {
    //分区内消息已保持有序
	messages.foreach(message => println(message))
	//返回true将自动提交分区内Offset
	true
  }).start()
}          

// 为Boots加载RabbitMQ插件  
import BootsRabbitMQ._

//消息发送
BootsDefault.rabbitMQ.sender()
	.config.sendRetryCount(1).build()
	.sendMessage("queueName", "123")
	
//消息消费
BootsDefault.rabbitMQ.consumer()
  .config.queueName("queueName").processorThreadCountMax(2).build()
  .process(message => {  
	println(message)
	// 返回true表示ACK成功
	true
  })
  .start()

@RestController
@RequestMapping(path = Array("/api/monitoring/project"))
class ProjectController extends RestApi {
	import web._
	
   // Action.Get	
   @GetMapping(Array("/application/list"))
   def application_list(request: HttpServletRequest) = sso(request) { implicit sessionContext =>
    ......
   }
   
   // Action.Post
   @PostMapping(Array("/application/set"))
   def application_set(request: HttpServletRequest) = sso(request) { implicit sessionContext =>
    ......
   }
}

val df = spark.read
  .format("nel.framework.boots.spark.client.source.hbase")
  .option("spark.cus.hbase.url","192.168.198.151")
  .option("spark.cus.hbase.table","dbo:test")
  .option("spark.cus.hbase.schema","c1 int,c2 long,c3 string") // row_key将被自动添加
  .option("spark.cus.hbase.start_key","00000001")
  .option("spark.cus.hbase.end_key","10000000")
  .load()

df.write
  .format("nel.framework.boots.spark.client.source.hbase")
  .option("spark.cus.hbase.url","192.168.198.151")
  .option("spark.cus.hbase.table","dbo:test")
  .option("spark.cus.hbase.tmp_path","hdfs://192.168.198.151:9000/xx/tmp/hbase_xxx")//MR数据落地Tmp地址
  //.option("spark.cus.hbase.row_key","row_key") // 默认df中row_key做为row_key,该列必须存在且有值   
  .save()

implicit val sparkConf = new SparkConf().setMaster("local[2]").setAppName("WCApp")

// SparkSQL  
BootsDefault.sparkClient
  // 关闭SparkServerMetric传输
  .config.isEnableSparkServer(false).build()
  .sql(spark => {
	val lines = spark.sparkContext.textFile("D:\\data\\test\\wc.txt")
	lines.flatMap(_.split(",")).map((_, 1))
	  .reduceByKey(_ + _)
	  .collect()
	  .foreach(println(_))
  });
  
// SparkStreaming
 BootsDefault.sparkClient
   .config.isEnableSparkServer(false).build()
   .streaming(Duration(5*1000)){ ssc =>
	 val lines = ssc.socketTextStream("xxx",9999)
	 //...
   }