pull/2148/head
Gordon
1 year ago
commit
a850733920
@ -1,68 +0,0 @@
|
||||
// Copyright © 2023 OpenIM. All rights reserved.
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package http
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/callbackstruct"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/config"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/servererrs"
|
||||
"github.com/openimsdk/tools/log"
|
||||
"github.com/openimsdk/tools/utils/httputil"
|
||||
)
|
||||
|
||||
var (
|
||||
// Define http client.
|
||||
client = httputil.NewHTTPClient(httputil.NewClientConfig())
|
||||
)
|
||||
|
||||
func init() {
|
||||
// reset http default transport
|
||||
http.DefaultTransport.(*http.Transport).MaxConnsPerHost = 100 // default: 2
|
||||
}
|
||||
|
||||
func callBackPostReturn(ctx context.Context, url, command string, input interface{}, output callbackstruct.CallbackResp, callbackConfig config.WebhookConfig) error {
|
||||
url = url + "/" + command
|
||||
log.ZInfo(ctx, "callback", "url", url, "input", input, "config", callbackConfig)
|
||||
b, err := client.Post(ctx, url, nil, input, callbackConfig.Timeout)
|
||||
if err != nil {
|
||||
if callbackConfig.FailedContinue {
|
||||
log.ZInfo(ctx, "callback failed but continue", err, "url", url)
|
||||
return nil
|
||||
}
|
||||
log.ZWarn(ctx, "callback network failed", err, "url", url, "input", input)
|
||||
return servererrs.ErrNetwork.WrapMsg(err.Error())
|
||||
}
|
||||
if err = json.Unmarshal(b, output); err != nil {
|
||||
if callbackConfig.FailedContinue {
|
||||
log.ZWarn(ctx, "callback failed but continue", err, "url", url)
|
||||
return nil
|
||||
}
|
||||
log.ZWarn(ctx, "callback json unmarshal failed", err, "url", url, "input", input, "response", string(b))
|
||||
return servererrs.ErrData.WithDetail(err.Error() + "response format error")
|
||||
}
|
||||
if err := output.Parse(); err != nil {
|
||||
log.ZWarn(ctx, "callback parse failed", err, "url", url, "input", input, "response", string(b))
|
||||
}
|
||||
log.ZInfo(ctx, "callback success", "url", url, "input", input, "response", string(b))
|
||||
return nil
|
||||
}
|
||||
|
||||
func CallBackPostReturn(ctx context.Context, url string, req callbackstruct.CallbackReq, resp callbackstruct.CallbackResp, callbackConfig config.WebhookConfig) error {
|
||||
return callBackPostReturn(ctx, url, req.GetCallbackCommand(), req, resp, callbackConfig)
|
||||
}
|
||||
@ -0,0 +1,74 @@
|
||||
// Copyright © 2023 OpenIM. All rights reserved.
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package webhook
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/callbackstruct"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/config"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/common/servererrs"
|
||||
"github.com/openimsdk/open-im-server/v3/pkg/util/memAsyncQueue"
|
||||
"github.com/openimsdk/protocol/constant"
|
||||
"github.com/openimsdk/tools/log"
|
||||
"github.com/openimsdk/tools/utils/httputil"
|
||||
"net/http"
|
||||
)
|
||||
|
||||
type Client struct {
|
||||
client *httputil.HTTPClient
|
||||
url string
|
||||
queue *memAsyncQueue.MemoryQueue
|
||||
}
|
||||
|
||||
func NewWebhookClient(url string, queue *memAsyncQueue.MemoryQueue) *Client {
|
||||
http.DefaultTransport.(*http.Transport).MaxConnsPerHost = 100 // Enhance the default number of max connections per host
|
||||
return &Client{
|
||||
client: httputil.NewHTTPClient(httputil.NewClientConfig()),
|
||||
url: url,
|
||||
queue: queue,
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Client) SyncPost(ctx context.Context, command string, req callbackstruct.CallbackReq, resp callbackstruct.CallbackResp, before *config.BeforeConfig) error {
|
||||
if before.Enable {
|
||||
return c.post(ctx, command, req, resp, before.Timeout)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *Client) AsyncPost(ctx context.Context, command string, req callbackstruct.CallbackReq, resp callbackstruct.CallbackResp, after *config.AfterConfig) {
|
||||
if after.Enable {
|
||||
c.queue.Push(func() { c.post(ctx, command, req, resp, after.Timeout) })
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Client) post(ctx context.Context, command string, input interface{}, output callbackstruct.CallbackResp, timeout int) error {
|
||||
fullURL := c.url + "/" + command
|
||||
log.ZInfo(ctx, "webhook", "url", fullURL, "input", input, "config", timeout)
|
||||
operationID, _ := ctx.Value(constant.OperationID).(string)
|
||||
b, err := c.client.Post(ctx, fullURL, map[string]string{constant.OperationID: operationID}, input, timeout)
|
||||
if err != nil {
|
||||
return servererrs.ErrNetwork.WrapMsg(err.Error(), "post url", fullURL)
|
||||
}
|
||||
if err = json.Unmarshal(b, output); err != nil {
|
||||
return servererrs.ErrData.WithDetail(err.Error() + " response format error")
|
||||
}
|
||||
if err := output.Parse(); err != nil {
|
||||
return err
|
||||
}
|
||||
log.ZInfo(ctx, "webhook success", "url", fullURL, "input", input, "response", string(b))
|
||||
return nil
|
||||
}
|
||||
@ -0,0 +1,74 @@
|
||||
package memAsyncQueue
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// AsyncQueue is the interface responsible for asynchronous processing of functions.
|
||||
type AsyncQueue interface {
|
||||
Initialize(processFunc func(), workerCount int, bufferSize int)
|
||||
Push(task func()) error
|
||||
}
|
||||
|
||||
// MemoryQueue is an implementation of the AsyncQueue interface using a channel to process functions.
|
||||
type MemoryQueue struct {
|
||||
taskChan chan func()
|
||||
wg sync.WaitGroup
|
||||
isStopped bool
|
||||
stopMutex sync.Mutex // Mutex to protect access to isStopped
|
||||
}
|
||||
|
||||
func NewMemoryQueue(workerCount int, bufferSize int) *MemoryQueue {
|
||||
mq := &MemoryQueue{} // Create a new instance of MemoryQueue
|
||||
mq.Initialize(workerCount, bufferSize) // Initialize it with specified parameters
|
||||
return mq
|
||||
}
|
||||
|
||||
// Initialize sets up the worker nodes and the buffer size of the channel,
|
||||
// starting internal goroutines to handle tasks from the channel.
|
||||
func (mq *MemoryQueue) Initialize(workerCount int, bufferSize int) {
|
||||
mq.taskChan = make(chan func(), bufferSize) // Initialize the channel with the provided buffer size.
|
||||
mq.isStopped = false
|
||||
|
||||
// Start multiple goroutines based on the specified workerCount.
|
||||
for i := 0; i < workerCount; i++ {
|
||||
mq.wg.Add(1)
|
||||
go func(workerID int) {
|
||||
defer mq.wg.Done()
|
||||
for task := range mq.taskChan {
|
||||
fmt.Printf("Worker %d: Executing task\n", workerID)
|
||||
task() // Execute the function
|
||||
}
|
||||
}(i)
|
||||
}
|
||||
}
|
||||
|
||||
// Push submits a function to the queue.
|
||||
// Returns an error if the queue is stopped or if the queue is full.
|
||||
func (mq *MemoryQueue) Push(task func()) error {
|
||||
mq.stopMutex.Lock()
|
||||
if mq.isStopped {
|
||||
mq.stopMutex.Unlock()
|
||||
return errors.New("push failed: queue is stopped")
|
||||
}
|
||||
mq.stopMutex.Unlock()
|
||||
|
||||
select {
|
||||
case mq.taskChan <- task:
|
||||
return nil
|
||||
case <-time.After(time.Millisecond * 100): // Timeout to prevent deadlock/blocking
|
||||
return errors.New("push failed: queue is full")
|
||||
}
|
||||
}
|
||||
|
||||
// Stop is used to terminate the internal goroutines and close the channel.
|
||||
func (mq *MemoryQueue) Stop() {
|
||||
mq.stopMutex.Lock()
|
||||
mq.isStopped = true
|
||||
close(mq.taskChan)
|
||||
mq.stopMutex.Unlock()
|
||||
mq.wg.Wait()
|
||||
}
|
||||
@ -0,0 +1,91 @@
|
||||
package memAsyncQueue
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// TestPushSuccess tests the successful pushing of data into the queue.
|
||||
func TestPushSuccess(t *testing.T) {
|
||||
queue := &MemoryQueue{}
|
||||
queue.Initialize(func(data any) {}, 1, 5) // Small buffer size for test
|
||||
|
||||
// Try to push data that should succeed
|
||||
err := queue.Push("test data")
|
||||
if err != nil {
|
||||
t.Errorf("Push should succeed, but got error: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestPushFailWhenFull tests that pushing to a full queue results in an error.
|
||||
func TestPushFailWhenFull(t *testing.T) {
|
||||
queue := &MemoryQueue{}
|
||||
queue.Initialize(func(data any) {
|
||||
time.Sleep(100 * time.Millisecond) // Simulate work to delay processing
|
||||
}, 1, 1) // Very small buffer to fill quickly
|
||||
|
||||
queue.Push("data 1") // Fill the buffer
|
||||
err := queue.Push("data 2") // This should fail
|
||||
|
||||
if err == nil {
|
||||
t.Error("Expected an error when pushing to full queue, but got none")
|
||||
}
|
||||
}
|
||||
|
||||
// TestPushFailWhenStopped tests that pushing to a stopped queue results in an error.
|
||||
func TestPushFailWhenStopped(t *testing.T) {
|
||||
queue := &MemoryQueue{}
|
||||
queue.Initialize(func(data any) {}, 1, 1)
|
||||
|
||||
queue.Stop() // Stop the queue before pushing
|
||||
err := queue.Push("test data")
|
||||
|
||||
if err == nil {
|
||||
t.Error("Expected an error when pushing to stopped queue, but got none")
|
||||
}
|
||||
}
|
||||
|
||||
// TestQueueOperationSequence tests a sequence of operations to ensure the queue handles them correctly.
|
||||
func TestQueueOperationSequence(t *testing.T) {
|
||||
queue := &MemoryQueue{}
|
||||
queue.Initialize(func(data any) {}, 1, 2)
|
||||
|
||||
// Sequence of pushes and a stop
|
||||
err := queue.Push("data 1")
|
||||
if err != nil {
|
||||
t.Errorf("Failed to push data 1: %v", err)
|
||||
}
|
||||
|
||||
err = queue.Push("data 2")
|
||||
if err != nil {
|
||||
t.Errorf("Failed to push data 2: %v", err)
|
||||
}
|
||||
|
||||
queue.Stop() // Stop the queue
|
||||
err = queue.Push("data 3") // This push should fail
|
||||
if err == nil {
|
||||
t.Error("Expected an error when pushing after stop, but got none")
|
||||
}
|
||||
}
|
||||
|
||||
// TestBlockingOnFull tests that the queue does not block indefinitely when full.
|
||||
func TestBlockingOnFull(t *testing.T) {
|
||||
queue := &MemoryQueue{}
|
||||
queue.Initialize(func(data any) {
|
||||
time.Sleep(1 * time.Second) // Simulate a long processing time
|
||||
}, 1, 1)
|
||||
|
||||
queue.Push("data 1") // Fill the queue
|
||||
|
||||
start := time.Now()
|
||||
err := queue.Push("data 2") // This should time out
|
||||
duration := time.Since(start)
|
||||
|
||||
if err == nil {
|
||||
t.Error("Expected an error due to full queue, but got none")
|
||||
}
|
||||
|
||||
if duration >= time.Second {
|
||||
t.Errorf("Push blocked for too long, duration: %v", duration)
|
||||
}
|
||||
}
|
||||
Loading…
Reference in new issue