github/goreplay
4
0
Fork 0
mirror of https://github.com/buger/goreplay.git synced 2026-10-20 00:16:42 +00:00
Code Issues Releases Wiki Activity

Fix issue after merging PRO code

Browse Source
This commit is contained in:
Leonid Bugaev
2024-10-25 15:05:42 +03:00
parent d3c95f155d
commit cca448bb9e
1 changed files with 15 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
output_http.go
+15 -15
View File
@@ -92,16 +92,16 @@ type httpWorker struct {
output *HTTPOutput
client *HTTPClient
lastActivity time.Time
queue chan []byte
queue chan *Message
stop chan bool
}
func newHTTPWorker(output *HTTPOutput, queue chan []byte) *httpWorker {
func newHTTPWorker(output *HTTPOutput, queue chan *Message) *httpWorker {
client := NewHTTPClient(output.config)
w := &httpWorker{client: client, output: output}
if queue == nil {
w.queue = make(chan []byte, 100)
w.queue = make(chan *Message, 100)
} else {
w.queue = queue
}
@@ -110,8 +110,8 @@ func newHTTPWorker(output *HTTPOutput, queue chan []byte) *httpWorker {
go func() {
for {
select {
case payload := <-w.queue:
output.sendRequest(client, payload)
case msg := <-w.queue:
output.sendRequest(client, msg)
case <-w.stop:
return
}
@@ -148,7 +148,7 @@ func NewHTTPOutput(address string, config *HTTPOutputConfig) PluginReadWriter {
newConfig.WorkersMin = 1000
}
if newConfig.WorkersMax <= 0 {
newConfig.WorkersMax = math.MaxInt32 // idealy so large
newConfig.WorkersMax = math.MaxInt32 // ideally so large
}
if newConfig.WorkersMax < newConfig.WorkersMin {
newConfig.WorkersMax = newConfig.WorkersMin
@@ -238,7 +238,7 @@ func (o *HTTPOutput) sessionWorkerMaster() {
o.workerSessions[sessionID] = worker
}
worker.queue <- msg.Data
worker.queue <- msg
worker.lastActivity = time.Now()
case <-gc:
now := time.Now()
@@ -260,7 +260,7 @@ func (o *HTTPOutput) startWorker() {
case <-o.stopWorker:
return
case msg := <-o.queue:
o.sendRequest(o.client, msg.Data)
o.sendRequest(o.client, msg)
}
}
}
@@ -321,14 +321,14 @@ func (o *HTTPOutput) PluginRead() (*Message, error) {
return &msg, nil
}
func (o *HTTPOutput) sendRequest(client *HTTPClient, data []byte) {
if !isRequestPayload(data) {
func (o *HTTPOutput) sendRequest(client *HTTPClient, msg *Message) {
if !isRequestPayload(msg.Meta) {
return
}
uuid := payloadID(data)
uuid := payloadID(msg.Meta)
start := time.Now()
resp, err := client.Send(data)
resp, err := client.Send(msg.Data)
stop := time.Now()
if err != nil {
@@ -344,7 +344,7 @@ func (o *HTTPOutput) sendRequest(client *HTTPClient, data []byte) {
}
if o.elasticSearch != nil {
o.elasticSearch.ResponseAnalyze(data, resp, start, stop)
o.elasticSearch.ResponseAnalyze(msg.Data, resp, start, stop)
}
}
@@ -384,7 +384,7 @@ func NewHTTPClient(config *HTTPOutputConfig) *HTTPClient {
},
}
if config.SkipVerify {
// clone to avoid modying global default RoundTripper
// clone to avoid modifying global default RoundTripper
transport = http.DefaultTransport.(*http.Transport).Clone()
transport.TLSClientConfig = &tls.Config{InsecureSkipVerify: true}
client.Client.Transport = transport
@@ -393,7 +393,7 @@ func NewHTTPClient(config *HTTPOutputConfig) *HTTPClient {
return client
}
// Send sends an http request using client create by NewHTTPClient
// Send sends an http request using client created by NewHTTPClient
func (c *HTTPClient) Send(data []byte) ([]byte, error) {
var req *http.Request
var resp *http.Response