-
Notifications
You must be signed in to change notification settings - Fork 16
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
56 changed files
with
973 additions
and
1,716 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,140 @@ | ||
| package kubemq | ||
|
|
||
| import ( | ||
| "context" | ||
| "encoding/json" | ||
| "fmt" | ||
| "github.com/kubemq-io/kubemq-go/common" | ||
| "github.com/kubemq-io/kubemq-go/pkg/uuid" | ||
| "time" | ||
| ) | ||
|
|
||
| const requestChannel = "kubemq.cluster.internal.requests" | ||
|
|
||
| func CreateChannel(ctx context.Context, client *Client, clientId string, channel string, channelType string) error { | ||
| request := NewQuery(). | ||
| SetChannel(requestChannel). | ||
| SetId(uuid.New()). | ||
| SetClientId(clientId). | ||
| SetMetadata("create-channel"). | ||
| SetTags(map[string]string{ | ||
| "channel_type": channelType, | ||
| "channel": channel, | ||
| "client_id": clientId, | ||
| }). | ||
| SetTimeout(time.Second * 10) | ||
| resp, err := client.SetQuery(request).Send(ctx) | ||
| if err != nil { | ||
| return fmt.Errorf("error sending create channel request: %s", err.Error()) | ||
| } | ||
| if resp.Error != "" { | ||
| return fmt.Errorf("error creating channel: %s", resp.Error) | ||
| } | ||
|
|
||
| return nil | ||
| } | ||
|
|
||
| func DeleteChannel(ctx context.Context, client *Client, clientId string, channel string, channelType string) error { | ||
| request := NewQuery(). | ||
| SetChannel(requestChannel). | ||
| SetId(uuid.New()). | ||
| SetClientId(clientId). | ||
| SetMetadata("delete-channel"). | ||
| SetTags(map[string]string{ | ||
| "channel_type": channelType, | ||
| "channel": channel, | ||
| "client_id": clientId, | ||
| }). | ||
| SetTimeout(time.Second * 10) | ||
| resp, err := client.SetQuery(request).Send(ctx) | ||
| if err != nil { | ||
| return fmt.Errorf("error sending delete channel request: %s", err.Error()) | ||
| } | ||
| if resp.Error != "" { | ||
| return fmt.Errorf("error deleting channel: %s", resp.Error) | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| func listChannels(ctx context.Context, client *Client, clientId string, channelType string, search string) ([]byte, error) { | ||
| request := NewQuery(). | ||
| SetChannel(requestChannel). | ||
| SetId(uuid.New()). | ||
| SetClientId(clientId). | ||
| SetMetadata("list-channels"). | ||
| SetTags(map[string]string{ | ||
| "channel_type": channelType, | ||
| "client_id": clientId, | ||
| "search": search, | ||
| }). | ||
| SetTimeout(time.Second * 10) | ||
| resp, err := client.SetQuery(request).Send(ctx) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("error sending list channels request: %s", err.Error()) | ||
| } | ||
| if resp.Error != "" { | ||
| return nil, fmt.Errorf("error listing channels: %s", resp.Error) | ||
| } | ||
| return resp.Body, nil | ||
| } | ||
| func ListQueuesChannels(ctx context.Context, client *Client, clientId string, search string) ([]*common.QueuesChannel, error) { | ||
| data, err := listChannels(ctx, client, clientId, "queues", search) | ||
| if err != nil { | ||
| return nil, err | ||
|
|
||
| } | ||
| return DecodeQueuesChannelList(data) | ||
| } | ||
|
|
||
| func ListPubSubChannels(ctx context.Context, client *Client, clientId string, channelType string, search string) ([]*common.PubSubChannel, error) { | ||
| data, err := listChannels(ctx, client, clientId, channelType, search) | ||
| if err != nil { | ||
| return nil, err | ||
|
|
||
| } | ||
| return DecodePubSubChannelList(data) | ||
| } | ||
|
|
||
| func ListCQChannels(ctx context.Context, client *Client, clientId string, channelType string, search string) ([]*common.CQChannel, error) { | ||
| data, err := listChannels(ctx, client, clientId, channelType, search) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| return DecodeCQChannelList(data) | ||
| } | ||
|
|
||
| func DecodePubSubChannelList(dataBytes []byte) ([]*common.PubSubChannel, error) { | ||
| var channelsData []*common.PubSubChannel | ||
| if dataBytes == nil { | ||
| return nil, nil | ||
| } | ||
| err := json.Unmarshal(dataBytes, &channelsData) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| return channelsData, nil | ||
| } | ||
|
|
||
| func DecodeQueuesChannelList(dataBytes []byte) ([]*common.QueuesChannel, error) { | ||
| var channelsData []*common.QueuesChannel | ||
| if dataBytes == nil { | ||
| return nil, nil | ||
| } | ||
| err := json.Unmarshal(dataBytes, &channelsData) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| return channelsData, nil | ||
| } | ||
|
|
||
| func DecodeCQChannelList(dataBytes []byte) ([]*common.CQChannel, error) { | ||
| var channelsData []*common.CQChannel | ||
| if dataBytes == nil { | ||
| return nil, nil | ||
| } | ||
| err := json.Unmarshal(dataBytes, &channelsData) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| return channelsData, nil | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,75 @@ | ||
| package common | ||
|
|
||
| import ( | ||
| "fmt" | ||
| ) | ||
|
|
||
| type QueuesStats struct { | ||
| Messages int `json:"messages"` | ||
| Volume int `json:"volume"` | ||
| Waiting int `json:"waiting"` | ||
| Expired int `json:"expired"` | ||
| Delayed int `json:"delayed"` | ||
| } | ||
|
|
||
| func (qs *QueuesStats) String() string { | ||
| return fmt.Sprintf("Stats: messages=%d, volume=%d, waiting=%d, expired=%d, delayed=%d", qs.Messages, qs.Volume, qs.Waiting, qs.Expired, qs.Delayed) | ||
| } | ||
|
|
||
| type QueuesChannel struct { | ||
| Name string `json:"name"` | ||
| Type string `json:"type"` | ||
| LastActivity int `json:"lastActivity"` | ||
| IsActive bool `json:"isActive"` | ||
| Incoming QueuesStats `json:"incoming"` | ||
| Outgoing QueuesStats `json:"outgoing"` | ||
| } | ||
|
|
||
| func (qc *QueuesChannel) String() string { | ||
| return fmt.Sprintf("Channel: name=%s, type=%s, last_activity=%d, is_active=%t, incoming=%v, outgoing=%v", qc.Name, qc.Type, qc.LastActivity, qc.IsActive, qc.Incoming, qc.Outgoing) | ||
| } | ||
|
|
||
| type PubSubStats struct { | ||
| Messages int `json:"messages"` | ||
| Volume int `json:"volume"` | ||
| } | ||
|
|
||
| func (ps *PubSubStats) String() string { | ||
| return fmt.Sprintf("Stats: messages=%d, volume=%d", ps.Messages, ps.Volume) | ||
| } | ||
|
|
||
| type PubSubChannel struct { | ||
| Name string `json:"name"` | ||
| Type string `json:"type"` | ||
| LastActivity int `json:"lastActivity"` | ||
| IsActive bool `json:"isActive"` | ||
| Incoming PubSubStats `json:"incoming"` | ||
| Outgoing PubSubStats `json:"outgoing"` | ||
| } | ||
|
|
||
| func (pc *PubSubChannel) String() string { | ||
| return fmt.Sprintf("Channel: name=%s, type=%s, last_activity=%d, is_active=%t, incoming=%v, outgoing=%v", pc.Name, pc.Type, pc.LastActivity, pc.IsActive, pc.Incoming, pc.Outgoing) | ||
| } | ||
|
|
||
| type CQStats struct { | ||
| Messages int `json:"messages"` | ||
| Volume int `json:"volume"` | ||
| Responses int `json:"responses"` | ||
| } | ||
|
|
||
| func (cs *CQStats) String() string { | ||
| return fmt.Sprintf("Stats: messages=%d, volume=%d, responses=%d", cs.Messages, cs.Volume, cs.Responses) | ||
| } | ||
|
|
||
| type CQChannel struct { | ||
| Name string `json:"name"` | ||
| Type string `json:"type"` | ||
| LastActivity int `json:"lastActivity"` | ||
| IsActive bool `json:"isActive"` | ||
| Incoming CQStats `json:"incoming"` | ||
| Outgoing CQStats `json:"outgoing"` | ||
| } | ||
|
|
||
| func (cc *CQChannel) String() string { | ||
| return fmt.Sprintf("Channel: name=%s, type=%s, last_activity=%d, is_active=%t, incoming=%v, outgoing=%v", cc.Name, cc.Type, cc.LastActivity, cc.IsActive, cc.Incoming, cc.Outgoing) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,28 @@ | ||
| package main | ||
|
|
||
| import ( | ||
| "context" | ||
| "github.com/kubemq-io/kubemq-go" | ||
| "log" | ||
| ) | ||
|
|
||
| func main() { | ||
| ctx, cancel := context.WithCancel(context.Background()) | ||
| defer cancel() | ||
| eventsStoreClient, err := kubemq.NewEventsStoreClient(ctx, | ||
| kubemq.WithAddress("localhost", 50000), | ||
| kubemq.WithClientId("example"), | ||
| kubemq.WithTransportType(kubemq.TransportTypeGRPC)) | ||
| if err != nil { | ||
| log.Fatal(err) | ||
| } | ||
| defer func() { | ||
| err := eventsStoreClient.Close() | ||
| if err != nil { | ||
| log.Fatal(err) | ||
| } | ||
| }() | ||
| if err := eventsStoreClient.Create(ctx, "events-store.single"); err != nil { | ||
| log.Fatal(err) | ||
| } | ||
| } |
Oops, something went wrong.