1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
|
// SPDX-License-Identifier: GPL-3.0-or-later
package activemq
import (
"encoding/xml"
"fmt"
"io"
"net/http"
"net/url"
"path"
"github.com/netdata/netdata/go/go.d.plugin/pkg/web"
)
type topics struct {
XMLName xml.Name `xml:"topics"`
Items []topic `xml:"topic"`
}
type topic struct {
XMLName xml.Name `xml:"topic"`
Name string `xml:"name,attr"`
Stats stats `xml:"stats"`
}
type queues struct {
XMLName xml.Name `xml:"queues"`
Items []queue `xml:"queue"`
}
type queue struct {
XMLName xml.Name `xml:"queue"`
Name string `xml:"name,attr"`
Stats stats `xml:"stats"`
}
type stats struct {
XMLName xml.Name `xml:"stats"`
Size int64 `xml:"size,attr"`
ConsumerCount int64 `xml:"consumerCount,attr"`
EnqueueCount int64 `xml:"enqueueCount,attr"`
DequeueCount int64 `xml:"dequeueCount,attr"`
}
const pathStats = "/%s/xml/%s.jsp"
func newAPIClient(client *http.Client, request web.Request, webadmin string) *apiClient {
return &apiClient{
httpClient: client,
request: request,
webadmin: webadmin,
}
}
type apiClient struct {
httpClient *http.Client
request web.Request
webadmin string
}
func (a *apiClient) getQueues() (*queues, error) {
req, err := a.createRequest(fmt.Sprintf(pathStats, a.webadmin, keyQueues))
if err != nil {
return nil, fmt.Errorf("error on creating request '%s' : %v", a.request.URL, err)
}
resp, err := a.doRequestOK(req)
defer closeBody(resp)
if err != nil {
return nil, err
}
var queues queues
if err := xml.NewDecoder(resp.Body).Decode(&queues); err != nil {
return nil, fmt.Errorf("error on decoding resp from %s : %s", req.URL, err)
}
return &queues, nil
}
func (a *apiClient) getTopics() (*topics, error) {
req, err := a.createRequest(fmt.Sprintf(pathStats, a.webadmin, keyTopics))
if err != nil {
return nil, fmt.Errorf("error on creating request '%s' : %v", a.request.URL, err)
}
resp, err := a.doRequestOK(req)
defer closeBody(resp)
if err != nil {
return nil, err
}
var topics topics
if err := xml.NewDecoder(resp.Body).Decode(&topics); err != nil {
return nil, fmt.Errorf("error on decoding resp from %s : %s", req.URL, err)
}
return &topics, nil
}
func (a *apiClient) doRequestOK(req *http.Request) (*http.Response, error) {
resp, err := a.httpClient.Do(req)
if err != nil {
return resp, fmt.Errorf("error on request to %s : %v", req.URL, err)
}
if resp.StatusCode != http.StatusOK {
return resp, fmt.Errorf("%s returned HTTP status %d", req.URL, resp.StatusCode)
}
return resp, err
}
func (a *apiClient) createRequest(urlPath string) (*http.Request, error) {
req := a.request.Copy()
u, err := url.Parse(req.URL)
if err != nil {
return nil, err
}
u.Path = path.Join(u.Path, urlPath)
req.URL = u.String()
return web.NewHTTPRequest(req)
}
func closeBody(resp *http.Response) {
if resp != nil && resp.Body != nil {
_, _ = io.Copy(io.Discard, resp.Body)
_ = resp.Body.Close()
}
}
|
