159 lines
3.3 KiB
Go
159 lines
3.3 KiB
Go
package i2es
|
|
|
|
import (
|
|
"bytes"
|
|
"encoding/json"
|
|
"errors"
|
|
"io/ioutil"
|
|
"net/http"
|
|
"strings"
|
|
|
|
log "github.com/Sirupsen/logrus"
|
|
"github.com/google/uuid"
|
|
)
|
|
|
|
// ESDoc Elasticsearch document structure
|
|
type ESDoc struct {
|
|
Echo string `json:"echo"`
|
|
Subg string `json:"subg"`
|
|
To string `json:"to"`
|
|
Author string `json:"author"`
|
|
Message string `json:"message"`
|
|
Date string `json:"date"`
|
|
MsgID string `json:"msgid"`
|
|
Tags string `json:"tags"`
|
|
Repto string `json:"repto"`
|
|
Address string `json:"address"`
|
|
TopicID string `json:"topicid"`
|
|
}
|
|
|
|
// ESConf ES connection settings
|
|
type ESConf struct {
|
|
Host string
|
|
Index string
|
|
Type string
|
|
}
|
|
|
|
type topicid struct{}
|
|
|
|
// findTopicID for msgid
|
|
func (t topicid) findTopicID(msgid string) (string, error) {
|
|
var topicid string
|
|
return topicid, nil
|
|
}
|
|
|
|
// check if
|
|
func (t topicid) checkID(msg *ESDoc) (bool, func()) {
|
|
if msg.TopicID != "" {
|
|
return false, func() {}
|
|
}
|
|
|
|
if msg.Repto == "" {
|
|
var genTopicID = func() {
|
|
u := uuid.New()
|
|
log.Debug("New UUID " + u.String() + " for message " + msg.MsgID)
|
|
msg.TopicID = u.String()
|
|
}
|
|
return true, genTopicID
|
|
}
|
|
|
|
var findAndSetTopicID = func() {
|
|
topicid, err := t.findTopicID(msg.MsgID)
|
|
if err != nil {
|
|
log.Error(err.Error())
|
|
}
|
|
log.Debug("Topic ID " + topicid + " found for message " + msg.MsgID)
|
|
}
|
|
return true, findAndSetTopicID
|
|
}
|
|
|
|
// getorcreate topicID. Generate new unique topicID if message is start message
|
|
// Find top message and get it topicid if ESDoc.Repto is not empty
|
|
func (t topicid) getOrCreate(msg *ESDoc) error {
|
|
// TODO: write the code
|
|
return nil
|
|
}
|
|
|
|
// PutToIndex ...
|
|
func (es ESConf) PutToIndex(msg ESDoc) error {
|
|
putURI := strings.Join([]string{es.Host, es.Index, es.Type, msg.MsgID}, "/")
|
|
log.Print(putURI)
|
|
|
|
doc, err := json.Marshal(msg)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
// log.Print(string(doc))
|
|
|
|
req, err := http.NewRequest("PUT", putURI, bytes.NewBuffer(doc))
|
|
req.Header.Add("Content-Type", "application/json")
|
|
|
|
client := &http.Client{}
|
|
resp, err := client.Do(req)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
defer resp.Body.Close()
|
|
|
|
return err
|
|
}
|
|
|
|
// ESRes ES response minimal structure
|
|
type ESRes struct {
|
|
Took int `json:"took"`
|
|
TimedOut bool `json:"timed_out"`
|
|
Hits Hits `json:"hits"`
|
|
}
|
|
|
|
// Hits ...
|
|
type Hits struct {
|
|
Total int `json:"total"`
|
|
MaxScore float32 `json:"max_score"`
|
|
Hits []interface{} `json:"hits"`
|
|
}
|
|
|
|
// CheckID ...
|
|
func (es ESConf) CheckID(id string) (bool, error) {
|
|
searchURI := strings.Join([]string{es.Host, es.Index, es.Type, "_search"}, "/")
|
|
|
|
searchQ := strings.Join([]string{`{"query": {"match": {"_id": "`, id, `"}}}`}, "")
|
|
|
|
req, err := http.NewRequest("POST", searchURI, bytes.NewBuffer([]byte(searchQ)))
|
|
req.Header.Add("Content-Type", "application/json")
|
|
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
|
|
client := &http.Client{}
|
|
resp, err := client.Do(req)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
|
|
defer resp.Body.Close()
|
|
|
|
body, err := ioutil.ReadAll(resp.Body)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
|
|
var e ESRes
|
|
err = json.Unmarshal(body, &e)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
|
|
if e.Hits.Total > 0 {
|
|
err = errors.New(strings.Join([]string{"Message ", id, " already in index"}, ""))
|
|
return false, err
|
|
}
|
|
if e.TimedOut {
|
|
err = errors.New("Request time out")
|
|
return false, err
|
|
}
|
|
|
|
return false, nil
|
|
}
|