elasticsearch/elasticsearch.go

113 lines
2.2 KiB
Go
Raw Normal View History

2022-03-10 04:01:06 +00:00
package elasticsearch
import (
"encoding/json"
"errors"
"fmt"
"github.com/Selly-Modules/natsio"
)
2022-03-10 04:36:06 +00:00
// Client ...
2022-03-10 04:01:06 +00:00
type Client struct {
Config Config
natsServer natsio.Server
natsJetStream natsio.JetStream
}
2022-03-10 04:36:06 +00:00
// NewClient
// Init client elasticsearch
2022-03-10 04:01:06 +00:00
func NewClient(config Config) (*Client, error) {
if config.ApiKey == "" {
return nil, errors.New("api key is required")
}
if config.Nats.URL == "" {
return nil, errors.New("nats url is required")
}
if err := natsio.Connect(config.Nats); err != nil {
return nil, fmt.Errorf("nats connect failed: %v", err)
}
c := &Client{
Config: config,
natsServer: natsio.GetServer(),
natsJetStream: natsio.GetJetStream(),
}
return c, nil
}
2022-03-10 04:36:06 +00:00
// SyncData
// Sync data to services ES
2022-03-10 04:01:06 +00:00
func (c *Client) SyncData(data SyncData) (bool, error) {
var (
res Response
2022-03-10 13:41:44 +00:00
req = RequestBody{
ApiKey: c.Config.ApiKey,
Body: toBytes(data),
}
2022-03-10 04:01:06 +00:00
)
2022-03-10 13:41:44 +00:00
msg, err := c.natsServer.Request(SubjectSyncData, toBytes(req))
2022-03-10 04:01:06 +00:00
if err != nil {
return false, err
}
if err = json.Unmarshal(msg.Data, &res); err != nil {
return false, err
}
if res.Message != "" {
return false, errors.New(res.Message)
}
return res.Success, nil
}
2022-03-10 04:36:06 +00:00
// Search
// Request search to service es
2022-03-10 13:41:44 +00:00
func (c *Client) Search(query ESQuery) (*Response, error) {
2022-03-10 04:01:06 +00:00
var (
2022-03-10 13:41:44 +00:00
req = RequestBody{
ApiKey: c.Config.ApiKey,
Body: toBytes(query),
}
res *Response
2022-03-10 04:01:06 +00:00
)
2022-03-10 13:41:44 +00:00
msg, err := c.natsServer.Request(SubjectSearch, toBytes(req))
2022-03-10 04:01:06 +00:00
if err != nil {
2022-03-10 13:41:44 +00:00
return nil, err
2022-03-10 04:01:06 +00:00
}
if err = json.Unmarshal(msg.Data, &res); err != nil {
2022-03-10 13:41:44 +00:00
return nil, err
2022-03-10 04:01:06 +00:00
}
if res.Message != "" {
2022-03-10 13:41:44 +00:00
return nil, errors.New(res.Message)
2022-03-10 04:01:06 +00:00
}
2022-03-10 13:41:44 +00:00
return res, nil
2022-03-10 04:01:06 +00:00
}
2022-03-10 04:36:06 +00:00
// UpdateDocument
// Insert or update document to ES
2022-03-10 13:41:44 +00:00
func (c *Client) UpdateDocument(query UpdateDataPayload) (bool, error) {
2022-03-10 04:01:06 +00:00
var (
2022-03-10 13:41:44 +00:00
req = RequestBody{
ApiKey: c.Config.ApiKey,
Body: toBytes(query),
}
2022-03-10 04:01:06 +00:00
res Response
)
2022-03-10 13:41:44 +00:00
msg, err := c.natsServer.Request(SubjectUpdateDocument, toBytes(req))
2022-03-10 04:01:06 +00:00
if err != nil {
2022-03-10 13:41:44 +00:00
return false, err
2022-03-10 04:01:06 +00:00
}
if err = json.Unmarshal(msg.Data, &res); err != nil {
2022-03-10 13:41:44 +00:00
return false, err
2022-03-10 04:01:06 +00:00
}
if res.Message != "" {
2022-03-10 13:41:44 +00:00
return res.Success, errors.New(res.Message)
2022-03-10 04:01:06 +00:00
}
2022-03-10 13:41:44 +00:00
return res.Success, nil
2022-03-10 04:01:06 +00:00
}
func toBytes(data interface{}) []byte {
b, _ := json.Marshal(data)
return b
}