[Update] Define method #5
|
@ -53,17 +53,17 @@ func GetClient() *Client {
|
|||
return client
|
||||
}
|
||||
|
||||
// RequestNats
|
||||
// requestNats
|
||||
// publish message to nats and waiting response
|
||||
func (c *Client) RequestNats(subject string, data []byte) (*Response, error) {
|
||||
func requestNats(subject string, data []byte) (*Response, error) {
|
||||
var (
|
||||
req = RequestBody{
|
||||
ApiKey: c.Config.ApiKey,
|
||||
ApiKey: client.Config.ApiKey,
|
||||
Body: data,
|
||||
}
|
||||
res *Response
|
||||
)
|
||||
msg, err := c.natsServer.Request(subject, toBytes(req))
|
||||
msg, err := client.natsServer.Request(subject, toBytes(req))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
@ -76,16 +76,16 @@ func (c *Client) RequestNats(subject string, data []byte) (*Response, error) {
|
|||
return res, nil
|
||||
}
|
||||
|
||||
// PublishWithJetStream
|
||||
// publishWithJetStream
|
||||
// Sync data to services ES with JetStream
|
||||
func (c *Client) PublishWithJetStream(streamName, subject string, data []byte) (bool, error) {
|
||||
func publishWithJetStream(streamName, subject string, data []byte) (bool, error) {
|
||||
var (
|
||||
req = RequestBody{
|
||||
ApiKey: c.Config.ApiKey,
|
||||
ApiKey: client.Config.ApiKey,
|
||||
Body: data,
|
||||
}
|
||||
)
|
||||
err := c.natsJetStream.Publish(streamName, subject, toBytes(req))
|
||||
err := client.natsJetStream.Publish(streamName, subject, toBytes(req))
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
|
|
6
pull.go
6
pull.go
|
@ -6,15 +6,15 @@ type Pull struct {
|
|||
|
||||
// ProductUpsert ...
|
||||
func (r Pull) ProductUpsert(query UpdateDataPayload) (bool, error) {
|
||||
return GetClient().PublishWithJetStream(JetStreamSearchService, SubjectPullProductUpsert, toBytes(query))
|
||||
return publishWithJetStream(JetStreamSearchService, SubjectPullProductUpsert, toBytes(query))
|
||||
}
|
||||
|
||||
// UserUpsert ...
|
||||
func (r Pull) UserUpsert(query UpdateDataPayload) (bool, error) {
|
||||
return GetClient().PublishWithJetStream(JetStreamSearchService, SubjectPullUserUpsert, toBytes(query))
|
||||
return publishWithJetStream(JetStreamSearchService, SubjectPullUserUpsert, toBytes(query))
|
||||
}
|
||||
|
||||
// OrderUpsert ...
|
||||
func (r Pull) OrderUpsert(query UpdateDataPayload) (bool, error) {
|
||||
return GetClient().PublishWithJetStream(JetStreamSearchService, SubjectPullOrderUpsert, toBytes(query))
|
||||
return publishWithJetStream(JetStreamSearchService, SubjectPullOrderUpsert, toBytes(query))
|
||||
}
|
||||
|
|
12
request.go
12
request.go
|
@ -6,30 +6,30 @@ type Request struct {
|
|||
|
||||
// ProductSearch ...
|
||||
func (r Request) ProductSearch(query ESQuery) (*Response, error) {
|
||||
return GetClient().RequestNats(SubjectRequestProductSearch, toBytes(query))
|
||||
return requestNats(SubjectRequestProductSearch, toBytes(query))
|
||||
}
|
||||
|
||||
// ProductUpsert ...
|
||||
func (r Request) ProductUpsert(query UpdateDataPayload) (*Response, error) {
|
||||
return GetClient().RequestNats(SubjectRequestProductUpsert, toBytes(query))
|
||||
return requestNats(SubjectRequestProductUpsert, toBytes(query))
|
||||
}
|
||||
|
||||
// UserSearch ...
|
||||
func (r Request) UserSearch(query ESQuery) (*Response, error) {
|
||||
return GetClient().RequestNats(SubjectRequestUserSearch, toBytes(query))
|
||||
return requestNats(SubjectRequestUserSearch, toBytes(query))
|
||||
}
|
||||
|
||||
// UserUpsert ...
|
||||
func (r Request) UserUpsert(query UpdateDataPayload) (*Response, error) {
|
||||
return GetClient().RequestNats(SubjectRequestUserUpsert, toBytes(query))
|
||||
return requestNats(SubjectRequestUserUpsert, toBytes(query))
|
||||
}
|
||||
|
||||
// OrderSearch ...
|
||||
func (r Request) OrderSearch(query ESQuery) (*Response, error) {
|
||||
return GetClient().RequestNats(SubjectRequestOrderSearch, toBytes(query))
|
||||
return requestNats(SubjectRequestOrderSearch, toBytes(query))
|
||||
}
|
||||
|
||||
// OrderUpsert ...
|
||||
func (r Request) OrderUpsert(query UpdateDataPayload) (*Response, error) {
|
||||
return GetClient().RequestNats(SubjectRequestOrderUpsert, toBytes(query))
|
||||
return requestNats(SubjectRequestOrderUpsert, toBytes(query))
|
||||
}
|
||||
|
|
Loading…
Reference in New Issue