fet: support batch message handler

@@ -24,15 +24,16 @@ type (
 	}
 	}
 
 
 	PostgresqlConf struct {
 	PostgresqlConf struct {
-		Host         string
-		Port         string
-		User      string
+		Host          string
+		Port          string
+		User          string
 		Password      string
 		Password      string
-		DBName 		  string
+		DBName        string
 		MaxQueueSize  int `json:",default=10000"`
 		MaxQueueSize  int `json:",default=10000"`
-		BatchSize  int `json:",default=100"`
+		BatchSize     int `json:",default=100"`
 		ThreadSize    int `json:",default=2"`
 		ThreadSize    int `json:",default=2"`
 		Interval      int `json:",default=5"`
 		Interval      int `json:",default=5"`
+		SleepInterval int `json:",default=2"`
 	}
 	}
 
 
 	Filter struct {
 	Filter struct {
@@ -63,7 +64,7 @@ type (
 		Filters []Filter `json:",optional"`
 		Filters []Filter `json:",optional"`
 		Output  struct {
 		Output  struct {
 			ElasticSearch ElasticSearchConf `json:",optional"`
 			ElasticSearch ElasticSearchConf `json:",optional"`
-			Postgresql   PostgresqlConf `json:",optional"`
+			Postgresql    PostgresqlConf    `json:",optional"`
 		}
 		}
 	}
 	}
 
 

@@ -35,13 +35,14 @@ func NewMessageHandlerPG(constant config.PostgresqlConf) *MessageHandlerPG {
 		db:         DB,
 		db:         DB,
 		conf:       constant,
 		conf:       constant,
 		queue:      collection.NewQueue(constant.MaxQueueSize),
 		queue:      collection.NewQueue(constant.MaxQueueSize),
-		limitCount: int32(float64(constant.MaxQueueSize) * 0.9),
+		limitCount: int32(float64(constant.MaxQueueSize) * 0.8),
 	}
 	}
 
 
 	for i := 0; i < constant.ThreadSize; i++ {
 	for i := 0; i < constant.ThreadSize; i++ {
-		go handler.TimerConsume()
+		thread :=i
+		go handler.TimerConsume(thread)
 	}
 	}
-	go handler.timerCreateLogTable()
+	go handler.TimerCreateLogTable()
 	return handler
 	return handler
 }
 }
 
 
@@ -63,54 +64,60 @@ func (mh *MessageHandlerPG) Consume(_, val string) error {
 	mh.queue.Put(m)
 	mh.queue.Put(m)
 	atomic.AddInt32(&mh.count, 1)
 	atomic.AddInt32(&mh.count, 1)
 	if mh.count > mh.limitCount {
 	if mh.count > mh.limitCount {
-		if _, ok := mh.queue.Take(); ok {
-			atomic.AddInt32(&mh.count, -1)
-		}
+		time.Sleep(time.Second * time.Duration(mh.conf.SleepInterval))
 	}
 	}
 	return nil
 	return nil
 }
 }
 
 
-func (mh *MessageHandlerPG) TimerConsume() {
+func (mh *MessageHandlerPG) TimerConsume(threadId int) {
+	fmt.Printf("[logstash] Begin TimerConsume : %v \n",threadId)
 	t := time.Tick(time.Second * time.Duration(mh.conf.Interval))
 	t := time.Tick(time.Second * time.Duration(mh.conf.Interval))
 	for range t {
 	for range t {
 		threading.RunSafe(
 		threading.RunSafe(
 			func() {
 			func() {
-				var logs []*Logs
-				for i := 0; i < mh.conf.BatchSize; i++ {
-					item, ok := mh.queue.Take()
-					if !ok {
-						break
+				for mh.count > 0 {
+					var logs []*Logs
+					for i := 0; i < mh.conf.BatchSize; i++ {
+						item, ok := mh.queue.Take()
+						if !ok || item == nil {
+							break
+						}
+						atomic.AddInt32(&mh.count, -1)
+						if m, ok := item.(map[string]interface{}); ok {
+							logs = append(logs, &Logs{
+								Log:     m,
+								LogTime: time.Now(),
+							})
+						}
 					}
 					}
-					atomic.AddInt32(&mh.count, -1)
-					if m, ok := item.(map[string]interface{}); ok {
-						logs = append(logs, &Logs{
-							Log:     m,
-							LogTime: time.Now(),
-						})
+					if len(logs) > 0 {
+						if _, err := mh.db.Model(&logs).Insert(); err != nil {
+							fmt.Println("[logstash] Insert Error:", err)
+						}
+					}
+					if mh.count>0{
+						fmt.Printf("[logstash] Thread:%v Queue:%v \n",threadId, mh.count)
 					}
 					}
 				}
 				}
-				if len(logs) > 0 {
-					_, err := mh.db.Model(&logs).Insert()
-					fmt.Println(err)
-				}
-				fmt.Printf("logstash thread:%v queue:%v \n", threading.RoutineId(), mh.count)
 			},
 			},
 		)
 		)
 	}
 	}
 }
 }
 
 
-func (mh *MessageHandlerPG) timerCreateLogTable() {
+func (mh *MessageHandlerPG) TimerCreateLogTable() {
 	t := time.NewTimer(time.Hour * 6)
 	t := time.NewTimer(time.Hour * 6)
+	fmt.Printf("[logstash] Begin TimerCreateLogTable \n")
+	mh.createLogTable()
 	for range t.C {
 	for range t.C {
 		threading.RunSafe(
 		threading.RunSafe(
 			func() {
 			func() {
-				mh.timerCreateLogTable()
+				mh.createLogTable()
 			},
 			},
 		)
 		)
 	}
 	}
 }
 }
 
 
-func (mh *MessageHandlerPG) TimerCreateLogTable() {
+func (mh *MessageHandlerPG) createLogTable() {
 	var err error
 	var err error
 	// creates database schema for Log models.
 	// creates database schema for Log models.
 	err = mh.db.Model(&Logs{}).CreateTable(&orm.CreateTableOptions{
 	err = mh.db.Model(&Logs{}).CreateTable(&orm.CreateTableOptions{
@@ -128,6 +135,7 @@ func (mh *MessageHandlerPG) TimerCreateLogTable() {
 		if err != nil {
 		if err != nil {
 			log.Fatal(err)
 			log.Fatal(err)
 		}
 		}
+
 		// logData := &Logs{
 		// logData := &Logs{
 		//     Log: map[string]interface{}{"msg":"test"},
 		//     Log: map[string]interface{}{"msg":"test"},
 		//     LogTime:   logStartTime,
 		//     LogTime:   logStartTime,
@@ -152,7 +160,7 @@ func createNewPartition(db *pg.DB, currentTime time.Time) error {
 		firstOfMonth.Format(time.RFC3339Nano),
 		firstOfMonth.Format(time.RFC3339Nano),
 		firstOfNextMonth.Format(time.RFC3339Nano),
 		firstOfNextMonth.Format(time.RFC3339Nano),
 	)
 	)
-
+	fmt.Println("[logstash] Create Partition:",sql)
 	_, err := db.Exec(sql)
 	_, err := db.Exec(sql)
 	return err
 	return err
 }
 }

@@ -80,8 +80,6 @@ func main() {
 		if len(processor.Output.Postgresql.Host) > 0 {
 		if len(processor.Output.Postgresql.Host) > 0 {
 			handle := handler.NewMessageHandlerPG(processor.Output.Postgresql)
 			handle := handler.NewMessageHandlerPG(processor.Output.Postgresql)
 			handle.AddFilters(filters...)
 			handle.AddFilters(filters...)
-			handle.TimerCreateLogTable()
-
 			
 			
 			for _, k := range toKqConf(processor.Input.Kafka) {
 			for _, k := range toKqConf(processor.Input.Kafka) {
 				group.Add(kq.MustNewQueue(k, handle))
 				group.Add(kq.MustNewQueue(k, handle))