Initialize LineUp app server
This commit is contained in:
@@ -0,0 +1,214 @@
|
||||
package wkapi
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
// The WuKongIM access API can briefly reject a foreground write while the
|
||||
// channel route converges to its current authority. A single client message
|
||||
// number is idempotent, so retrying the same request is safe.
|
||||
var sendRetryDelays = []time.Duration{
|
||||
50 * time.Millisecond,
|
||||
100 * time.Millisecond,
|
||||
200 * time.Millisecond,
|
||||
400 * time.Millisecond,
|
||||
}
|
||||
|
||||
// Client wraps HTTP calls to WuKongIM 3.0 API.
|
||||
type Client struct {
|
||||
apiURL string
|
||||
cli *http.Client
|
||||
}
|
||||
|
||||
// New creates a new WuKongIM API client.
|
||||
func New(apiURL string) *Client {
|
||||
return &Client{apiURL: apiURL, cli: &http.Client{}}
|
||||
}
|
||||
|
||||
// TokenResponse is the response from POST /user/token.
|
||||
type TokenResponse struct {
|
||||
UID string `json:"uid"`
|
||||
}
|
||||
|
||||
// RegisterToken registers a user token with WuKongIM.
|
||||
func (c *Client) RegisterToken(uid, token string) error {
|
||||
body := map[string]interface{}{
|
||||
"uid": uid,
|
||||
"token": token,
|
||||
"device_flag": 0,
|
||||
"device_level": 1,
|
||||
}
|
||||
var resp TokenResponse
|
||||
return c.post("/user/token", body, &resp)
|
||||
}
|
||||
|
||||
// UpsertChannel creates or updates a channel.
|
||||
func (c *Client) UpsertChannel(channelID string, channelType uint8, subscribers []string) error {
|
||||
body := map[string]interface{}{
|
||||
"channel_id": channelID,
|
||||
"channel_type": channelType,
|
||||
}
|
||||
if len(subscribers) > 0 {
|
||||
body["subscribers"] = subscribers
|
||||
}
|
||||
return c.post("/channel", body, nil)
|
||||
}
|
||||
|
||||
// SubscriberAdd adds subscribers to a channel.
|
||||
func (c *Client) SubscriberAdd(channelID string, channelType uint8, uids []string) error {
|
||||
body := map[string]interface{}{
|
||||
"channel_id": channelID,
|
||||
"channel_type": channelType,
|
||||
"reset": 0,
|
||||
"subscribers": uids,
|
||||
}
|
||||
return c.post("/channel/subscriber_add", body, nil)
|
||||
}
|
||||
|
||||
// SendMessageRequest is the request body for POST /message/send.
|
||||
type SendMessageRequest struct {
|
||||
FromUID string `json:"from_uid"`
|
||||
ChannelID string `json:"channel_id"`
|
||||
ChannelType uint8 `json:"channel_type"`
|
||||
ClientMsgNo string `json:"client_msg_no"`
|
||||
Payload string `json:"payload"` // base64-encoded
|
||||
}
|
||||
|
||||
// SendMessageResponse is the response from POST /message/send.
|
||||
type SendMessageResponse struct {
|
||||
MessageID int64 `json:"message_id"`
|
||||
MessageSeq uint64 `json:"message_seq"`
|
||||
Reason uint8 `json:"reason"`
|
||||
}
|
||||
|
||||
// HTTPStatusError retains an API failure status so callers can distinguish a
|
||||
// transient route response from a permanent validation or permission error.
|
||||
type HTTPStatusError struct {
|
||||
StatusCode int
|
||||
Body string
|
||||
}
|
||||
|
||||
func (e *HTTPStatusError) Error() string {
|
||||
return fmt.Sprintf("WK API returned status %d: %s", e.StatusCode, e.Body)
|
||||
}
|
||||
|
||||
// SendMessage sends a message via WuKongIM.
|
||||
func (c *Client) SendMessage(req SendMessageRequest) (*SendMessageResponse, error) {
|
||||
for attempt := 0; ; attempt++ {
|
||||
var resp SendMessageResponse
|
||||
err := c.post("/message/send", req, &resp)
|
||||
if err == nil {
|
||||
return &resp, nil
|
||||
}
|
||||
if !isRetryRequired(err) || attempt >= len(sendRetryDelays) {
|
||||
return nil, err
|
||||
}
|
||||
time.Sleep(sendRetryDelays[attempt])
|
||||
}
|
||||
}
|
||||
|
||||
func isRetryRequired(err error) bool {
|
||||
var statusErr *HTTPStatusError
|
||||
return errors.As(err, &statusErr) &&
|
||||
statusErr.StatusCode == http.StatusServiceUnavailable &&
|
||||
strings.Contains(statusErr.Body, `"retry required"`)
|
||||
}
|
||||
|
||||
// SyncChannelMessagesRequest is the request body for POST /channel/messagesync.
|
||||
type SyncChannelMessagesRequest struct {
|
||||
LoginUID string `json:"login_uid"`
|
||||
ChannelID string `json:"channel_id"`
|
||||
ChannelType uint8 `json:"channel_type"`
|
||||
StartMessageSeq uint64 `json:"start_message_seq"`
|
||||
Limit int `json:"limit"`
|
||||
PullMode int `json:"pull_mode"`
|
||||
}
|
||||
|
||||
// LegacyMessageResp is a message in the sync response (v3 legacy format).
|
||||
type LegacyMessageResp struct {
|
||||
Header map[string]interface{} `json:"header"`
|
||||
Setting uint8 `json:"setting"`
|
||||
MessageID uint64 `json:"message_id"`
|
||||
ClientMsgNo string `json:"client_msg_no"`
|
||||
MessageSeq uint64 `json:"message_seq"`
|
||||
FromUID string `json:"from_uid"`
|
||||
ChannelID string `json:"channel_id"`
|
||||
ChannelType uint8 `json:"channel_type"`
|
||||
Timestamp int32 `json:"timestamp"`
|
||||
Payload string `json:"payload"` // base64-encoded
|
||||
}
|
||||
|
||||
// SyncChannelMessagesResponse is the response from POST /channel/messagesync.
|
||||
type SyncChannelMessagesResponse struct {
|
||||
StartMessageSeq uint64 `json:"start_message_seq"`
|
||||
EndMessageSeq uint64 `json:"end_message_seq"`
|
||||
More int `json:"more"`
|
||||
Messages []LegacyMessageResp `json:"messages"`
|
||||
}
|
||||
|
||||
// SyncChannelMessages syncs messages from a channel.
|
||||
func (c *Client) SyncChannelMessages(req SyncChannelMessagesRequest) (*SyncChannelMessagesResponse, error) {
|
||||
var resp SyncChannelMessagesResponse
|
||||
if err := c.post("/channel/messagesync", req, &resp); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &resp, nil
|
||||
}
|
||||
|
||||
// RouteResponse is the response from GET /route.
|
||||
type RouteResponse struct {
|
||||
TCPAddr string `json:"tcp_addr"`
|
||||
WSAddr string `json:"ws_addr"`
|
||||
WSSAddr string `json:"wss_addr"`
|
||||
}
|
||||
|
||||
// GetRoute gets the WebSocket route from WuKongIM.
|
||||
func (c *Client) GetRoute() (*RouteResponse, error) {
|
||||
var resp RouteResponse
|
||||
if err := c.get("/route", &resp); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &resp, nil
|
||||
}
|
||||
|
||||
// --- internal helpers ---
|
||||
|
||||
func (c *Client) post(path string, body interface{}, dest interface{}) error {
|
||||
b, err := json.Marshal(body)
|
||||
if err != nil {
|
||||
return fmt.Errorf("marshal request: %w", err)
|
||||
}
|
||||
resp, err := c.cli.Post(c.apiURL+path, "application/json", bytes.NewReader(b))
|
||||
if err != nil {
|
||||
return fmt.Errorf("http post %s: %w", path, err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
return c.decodeResponse(resp, dest)
|
||||
}
|
||||
|
||||
func (c *Client) get(path string, dest interface{}) error {
|
||||
resp, err := c.cli.Get(c.apiURL + path)
|
||||
if err != nil {
|
||||
return fmt.Errorf("http get %s: %w", path, err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
return c.decodeResponse(resp, dest)
|
||||
}
|
||||
|
||||
func (c *Client) decodeResponse(resp *http.Response, dest interface{}) error {
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
body, _ := io.ReadAll(resp.Body)
|
||||
return &HTTPStatusError{StatusCode: resp.StatusCode, Body: string(body)}
|
||||
}
|
||||
if dest == nil {
|
||||
return nil
|
||||
}
|
||||
return json.NewDecoder(resp.Body).Decode(dest)
|
||||
}
|
||||
@@ -0,0 +1,69 @@
|
||||
package wkapi
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestSendMessageRetriesRetryRequiredWithSameClientMessageNumber(t *testing.T) {
|
||||
var attempts atomic.Int32
|
||||
var clientMsgNo string
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if r.URL.Path != "/message/send" {
|
||||
t.Fatalf("path = %s, want /message/send", r.URL.Path)
|
||||
}
|
||||
var got SendMessageRequest
|
||||
if err := json.NewDecoder(r.Body).Decode(&got); err != nil {
|
||||
t.Fatalf("decode request: %v", err)
|
||||
}
|
||||
if clientMsgNo == "" {
|
||||
clientMsgNo = got.ClientMsgNo
|
||||
} else if got.ClientMsgNo != clientMsgNo {
|
||||
t.Fatalf("client_msg_no = %q, want stable %q", got.ClientMsgNo, clientMsgNo)
|
||||
}
|
||||
if attempts.Add(1) <= 2 {
|
||||
w.WriteHeader(http.StatusServiceUnavailable)
|
||||
_, _ = w.Write([]byte(`{"error":"retry required"}`))
|
||||
return
|
||||
}
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
_, _ = w.Write([]byte(`{"message_id":101,"message_seq":7,"reason":1}`))
|
||||
}))
|
||||
defer server.Close()
|
||||
|
||||
client := New(server.URL)
|
||||
response, err := client.SendMessage(SendMessageRequest{
|
||||
FromUID: "user_1", ChannelID: "agent_channel", ChannelType: 2,
|
||||
ClientMsgNo: "lineup-idempotent-message", Payload: "aGVsbG8=",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("SendMessage() error = %v", err)
|
||||
}
|
||||
if response.MessageID != 101 || response.MessageSeq != 7 {
|
||||
t.Fatalf("SendMessage() response = %#v", response)
|
||||
}
|
||||
if got := attempts.Load(); got != 3 {
|
||||
t.Fatalf("attempts = %d, want 3", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSendMessageDoesNotRetryPermanentAPIError(t *testing.T) {
|
||||
var attempts atomic.Int32
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
attempts.Add(1)
|
||||
w.WriteHeader(http.StatusBadRequest)
|
||||
_, _ = w.Write([]byte(`{"error":"invalid request"}`))
|
||||
}))
|
||||
defer server.Close()
|
||||
|
||||
_, err := New(server.URL).SendMessage(SendMessageRequest{ClientMsgNo: "invalid"})
|
||||
if err == nil {
|
||||
t.Fatal("SendMessage() error = nil, want permanent error")
|
||||
}
|
||||
if got := attempts.Load(); got != 1 {
|
||||
t.Fatalf("attempts = %d, want 1", got)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user