diff --git a/README.ko-KR.md b/README.ko-KR.md index 32ba299..9534ebe 100644 --- a/README.ko-KR.md +++ b/README.ko-KR.md @@ -11,6 +11,7 @@

Go Version + Website License: MIT Platform Architecture @@ -76,10 +77,10 @@ flowchart LR | Broker Type | 버전 | TUI 대시보드 (Jolokia) | 메시지 송수신 (AMQP/STOMP) | 비고 | | :--- | :--- | :---: | :---: | :--- | -| **ActiveMQ Classic** | 5.x ~ 6.x | 🟢 완벽 지원 | 🟢 완벽 지원 | `amqcli`의 주 타겟 브로커입니다. | -| **ActiveMQ Artemis** | 2.x ~ | ❌ 미지원 | 🟢 지원 | Artemis는 JMX MBean 구조가 완전히 달라 대시보드는 렌더링되지 않으나, 프로토콜 기반 송수신은 가능합니다. | +| **ActiveMQ Classic** | 5.x ~ 6.x | 🟢 완벽 지원 | 🟢 완벽 지원 | Native JMX/Jolokia MBean 기반 관리 지원 | +| **ActiveMQ Artemis** | 2.x ~ 3.x | 🚧 작업 중 (WIP) | 🚧 작업 중 (WIP) | 현재 개발 작업이 진행 중입니다. | -> *참고: `amqcli`의 TUI 대시보드가 보여주는 Jolokia(JMX) 메트릭은 브로커 메모리 상의 실시간 뷰(Runtime View)입니다. KahaDB 등 디스크에 저장된 영속성(Persistence) 데이터 전체와 미세한 시차가 발생할 수 있습니다.* +> *참고: `amqcli`는 브로커에 연결할 때 대상이 Classic인지 Artemis인지 자동으로 탐색하여 최적의 어댑터를 연결합니다.* --- @@ -100,12 +101,12 @@ brew install amqcli **macOS / Linux (Shell)** ```bash -curl -fsSL https://raw.githubusercontent.com/xvlet/amqcli/master/install.sh | sh +curl -fsSL https://amqcli.pages.dev/install.sh | sh ``` **Windows (PowerShell)** ```powershell -powershell -ExecutionPolicy Bypass -c "irm https://raw.githubusercontent.com/xvlet/amqcli/master/install.ps1 | iex" +powershell -ExecutionPolicy Bypass -c "irm https://amqcli.pages.dev/install.ps1 | iex" ``` ### 3. Go 명령어 사용 (go install) @@ -146,7 +147,8 @@ refresh_interval: 1s encoding: utf-8 environments: dev: - protocol: "stomp" # 또는 "amqp" + broker_type: "auto" # "auto", "classic", 또는 "artemis" (기본값: auto) + protocol: "stomp" # 또는 "amqp" host: "${MQ_HOST:-127.0.0.1}" stomp_port: "61613" # 선택 사항 (기본값: 61613) web_port: "8161" # 선택 사항 (기본값: 8161) @@ -154,6 +156,7 @@ environments: password: "${MQ_PASS:-admin}" readonly: false prod: + broker_type: "auto" protocol: "amqp" host: "10.0.0.5" amqp_port: "5672" # 선택 사항 (기본값: 5672) diff --git a/README.md b/README.md index d72067a..d28686b 100644 --- a/README.md +++ b/README.md @@ -11,6 +11,7 @@

Go Version + Website License: MIT Platform Architecture @@ -76,10 +77,10 @@ flowchart LR | Broker Type | Versions | TUI Dashboard (Jolokia) | Messaging (AMQP/STOMP) | Notes | | :--- | :--- | :---: | :---: | :--- | -| **ActiveMQ Classic** | 5.x ~ 6.x | 🟢 Fully Supported | 🟢 Fully Supported | The primary target broker for `amqcli`. | -| **ActiveMQ Artemis** | 2.x ~ | ❌ Not Supported | 🟢 Supported | Artemis uses a completely different JMX MBean structure. While the TUI dashboard cannot render metrics, protocol-based message sending still functions. | +| **ActiveMQ Classic** | 5.x ~ 6.x | 🟢 Fully Supported | 🟢 Fully Supported | Native JMX/Jolokia MBean management. | +| **ActiveMQ Artemis** | 2.x ~ 3.x | 🚧 Work In Progress | 🚧 Work In Progress | Development in progress. | -> *Note: The Jolokia (JMX) metrics displayed in the TUI represent the broker's real-time runtime memory view, which may have slight discrepancies with the fully persisted KahaDB disk state.* +> *Note: `amqcli` automatically detects whether the target broker is Classic or Artemis. The Jolokia (JMX) metrics displayed in the TUI represent the broker's real-time runtime memory view.* --- @@ -100,12 +101,12 @@ The easiest way to install the latest release is by using the provided installat **macOS / Linux (Shell)** ```bash -curl -fsSL https://raw.githubusercontent.com/xvlet/amqcli/master/install.sh | sh +curl -fsSL https://amqcli.pages.dev/install.sh | sh ``` **Windows (PowerShell)** ```powershell -powershell -ExecutionPolicy Bypass -c "irm https://raw.githubusercontent.com/xvlet/amqcli/master/install.ps1 | iex" +powershell -ExecutionPolicy Bypass -c "irm https://amqcli.pages.dev/install.ps1 | iex" ``` ### 3. Using Go (go install) @@ -146,7 +147,8 @@ refresh_interval: 1s encoding: utf-8 environments: dev: - protocol: "stomp" # or "amqp" + broker_type: "auto" # "auto", "classic", or "artemis" (default: auto) + protocol: "stomp" # "stomp" or "amqp" host: "${MQ_HOST:-127.0.0.1}" stomp_port: "61613" # optional (default: 61613) web_port: "8161" # optional (default: 8161) @@ -154,6 +156,7 @@ environments: password: "${MQ_PASS:-admin}" readonly: false prod: + broker_type: "auto" protocol: "amqp" host: "10.0.0.5" amqp_port: "5672" # optional (default: 5672) diff --git a/adapter/inbound/ui/tui.go b/adapter/inbound/ui/tui.go index 868a9e7..45a3ccf 100644 --- a/adapter/inbound/ui/tui.go +++ b/adapter/inbound/ui/tui.go @@ -139,11 +139,11 @@ func NewAppModel(uc usecase.UseCase, interval time.Duration, host string, env st // 1. Queue Table qCols := []table.Column{ - {Title: "Name", Width: 26}, - {Title: fmt.Sprintf("%10s", "Pending"), Width: 10}, - {Title: fmt.Sprintf("%10s", "Consumers"), Width: 10}, - {Title: fmt.Sprintf("%10s", "Enqueued"), Width: 10}, - {Title: fmt.Sprintf("%10s", "Dequeued"), Width: 10}, + {Title: "Name", Width: 46}, + {Title: fmt.Sprintf("%12s", "Pending"), Width: 12}, + {Title: fmt.Sprintf("%12s", "Consumers"), Width: 12}, + {Title: fmt.Sprintf("%12s", "Enqueued"), Width: 12}, + {Title: fmt.Sprintf("%12s", "Dequeued"), Width: 12}, } qTable := table.New(table.WithColumns(qCols), table.WithFocused(true)) diff --git a/adapter/inbound/ui/tui_update.go b/adapter/inbound/ui/tui_update.go index 2633675..31a58ac 100644 --- a/adapter/inbound/ui/tui_update.go +++ b/adapter/inbound/ui/tui_update.go @@ -779,26 +779,8 @@ func (m *AppModel) updateConnections(msg tea.Msg) (tea.Model, tea.Cmd) { } func (m *AppModel) updateQueueTableColumns() { - // Name width is preserved from existing configuration to avoid shrinking unexpectedly - qNameW := 26 - if len(m.queueTable.Columns()) > 0 { - qNameW = m.queueTable.Columns()[0].Width - } - qCols := []table.Column{ - {Title: "Name", Width: qNameW}, - {Title: fmt.Sprintf("%12s", "Pending"), Width: 12}, - {Title: fmt.Sprintf("%12s", "Consumers"), Width: 12}, - {Title: fmt.Sprintf("%12s", "Enqueued"), Width: 12}, - {Title: fmt.Sprintf("%12s", "Dequeued"), Width: 12}, - } - if m.viewStats { - qCols = append(qCols, - table.Column{Title: "Memory", Width: 30}, - table.Column{Title: "Disk", Width: 15}, - ) - } m.queueTable.SetRows([]table.Row{}) // Temporarily clear rows to prevent panic during SetColumns - m.queueTable.SetColumns(qCols) + m.recalculateTableWidths() } func (m *AppModel) buildQueueRows(queues []domain.Queue) []table.Row { diff --git a/adapter/inbound/ui/tui_util.go b/adapter/inbound/ui/tui_util.go index 3991732..2904160 100644 --- a/adapter/inbound/ui/tui_util.go +++ b/adapter/inbound/ui/tui_util.go @@ -14,44 +14,69 @@ func (m *AppModel) recalculateTableWidths() { return } - // Calculate inner width for safety checks (Margin(2)+Border(2)+Padding(2)=6 offset) - contentWidth := m.width - 10 + // Box inner usable width: Width(m.width-6) with Border(2) and Padding(2) gives m.width - 10 + innerUsableWidth := m.width - 10 + if innerUsableWidth < 40 { + innerUsableWidth = 40 + } // 1. Queue Table Resizing - // Use original fixed widths, only shrink Name if terminal is too narrow - qNameW := 26 - if contentWidth < 105 { - qNameW = contentWidth - 70 + // Max cap at 46 so UUID queue names (36 chars) fit completely without creating huge empty gaps in fullscreen + if !m.viewStats { + // 5 columns: Name + 4 fixed metrics (12 chars each = 48) + 5*2 cell padding (10) = 58 fixed overhead + qNameW := innerUsableWidth - 58 - 2 + if qNameW > 46 { + qNameW = 46 + } if qNameW < 10 { qNameW = 10 } - } - qCols := []table.Column{ - {Title: "Name", Width: qNameW}, - {Title: fmt.Sprintf("%12s", "Pending"), Width: 12}, - {Title: fmt.Sprintf("%12s", "Consumers"), Width: 12}, - {Title: fmt.Sprintf("%12s", "Enqueued"), Width: 12}, - {Title: fmt.Sprintf("%12s", "Dequeued"), Width: 12}, - } - if m.viewStats { - qCols = append(qCols, - table.Column{Title: "Memory", Width: 30}, - table.Column{Title: "Disk", Width: 15}, - ) - } - m.queueTable.SetColumns(qCols) - - // 2. Message Table Resizing - // Use original fixed widths where possible, shrink ID/Correlation proportionally if needed - mIdW := 46 - mCorrW := 36 - if contentWidth < 164 { - slack := contentWidth - 82 - if slack < 20 { - slack = 20 + m.queueTable.SetColumns([]table.Column{ + {Title: "Name", Width: qNameW}, + {Title: fmt.Sprintf("%12s", "Pending"), Width: 12}, + {Title: fmt.Sprintf("%12s", "Consumers"), Width: 12}, + {Title: fmt.Sprintf("%12s", "Enqueued"), Width: 12}, + {Title: fmt.Sprintf("%12s", "Dequeued"), Width: 12}, + }) + } else { + // 7 columns: Name + 4 metrics (48) + Memory (26) + Disk (12) = 86 + 7*2 padding (14) = 100 fixed overhead + qNameW := innerUsableWidth - 100 - 2 + if qNameW > 46 { + qNameW = 46 + } + if qNameW < 10 { + qNameW = 10 } - mIdW = int(float64(slack) * 0.56) - mCorrW = slack - mIdW + m.queueTable.SetColumns([]table.Column{ + {Title: "Name", Width: qNameW}, + {Title: fmt.Sprintf("%12s", "Pending"), Width: 12}, + {Title: fmt.Sprintf("%12s", "Consumers"), Width: 12}, + {Title: fmt.Sprintf("%12s", "Enqueued"), Width: 12}, + {Title: fmt.Sprintf("%12s", "Dequeued"), Width: 12}, + {Title: "Memory", Width: 26}, + {Title: "Disk", Width: 12}, + }) + } + + // 2. Message Table Resizing (8 columns) + // 8*2 padding (16) + SEQ(5) + Persistence(12) + Priority(8) + Redelivered(12) + Timestamp(24) + Action(10) = 87 fixed overhead + msgSlack := innerUsableWidth - 87 - 2 + if msgSlack < 20 { + msgSlack = 20 + } + mIdW := int(float64(msgSlack) * 0.55) + mCorrW := msgSlack - mIdW + if mIdW > 46 { + mIdW = 46 + } + if mCorrW > 36 { + mCorrW = 36 + } + if mIdW < 10 { + mIdW = 10 + } + if mCorrW < 10 { + mCorrW = 10 } m.msgTable.SetColumns([]table.Column{ @@ -61,23 +86,31 @@ func (m *AppModel) recalculateTableWidths() { {Title: "Persistence", Width: 12}, {Title: "Priority", Width: 8}, {Title: "Redelivered", Width: 12}, - {Title: "Timestamp", Width: 30}, - {Title: "Action", Width: 15}, + {Title: "Timestamp", Width: 24}, + {Title: "Action", Width: 10}, }) - // 3. Connections Table Resizing - cNameW := 40 - cAddrW := 30 - if contentWidth < 90 { - cAddrW = contentWidth - 60 - if cAddrW < 10 { - cAddrW = 10 - } - cNameW = contentWidth - cAddrW - 20 - if cNameW < 10 { - cNameW = 10 - } + // 3. Connections Table Resizing (4 columns) + // 4*2 padding (8) + Active(10) + Slow(10) = 28 fixed overhead + connSlack := innerUsableWidth - 28 - 2 + if connSlack < 20 { + connSlack = 20 + } + cNameW := int(float64(connSlack) * 0.55) + cAddrW := connSlack - cNameW + if cNameW > 46 { + cNameW = 46 + } + if cAddrW > 32 { + cAddrW = 32 + } + if cNameW < 10 { + cNameW = 10 } + if cAddrW < 10 { + cAddrW = 10 + } + m.connectionsTable.SetColumns([]table.Column{ {Title: "Name", Width: cNameW}, {Title: "Remote Address", Width: cAddrW}, @@ -85,31 +118,22 @@ func (m *AppModel) recalculateTableWidths() { {Title: "Slow", Width: 10}, }) - // 4. Consumers Table Resizing - conPidW := 10 - conAddrW := 25 - conClientW := 40 - conDeqW := 10 - conUptimeW := 15 - - if contentWidth < 100 { - // narrower terminal, shrink ClientID proportionally - conClientW = contentWidth - 60 - if conClientW < 10 { - conClientW = 10 - } - conAddrW = contentWidth - conClientW - 35 - if conAddrW < 10 { - conAddrW = 10 - } + // 4. Consumers Table Resizing (5 columns) + // 5*2 padding (10) + PID(10) + Remote Address(22) + Dequeues(10) + Uptime(12) = 64 fixed overhead + conClientW := innerUsableWidth - 64 - 2 + if conClientW > 46 { + conClientW = 46 + } + if conClientW < 10 { + conClientW = 10 } m.consumersTable.SetColumns([]table.Column{ - {Title: "PID", Width: conPidW}, - {Title: "Remote Address", Width: conAddrW}, + {Title: "PID", Width: 10}, + {Title: "Remote Address", Width: 22}, {Title: "Client ID", Width: conClientW}, - {Title: "Dequeues", Width: conDeqW}, - {Title: "Uptime", Width: conUptimeW}, + {Title: "Dequeues", Width: 10}, + {Title: "Uptime", Width: 12}, }) } diff --git a/adapter/outbound/activemq/artemis_jolokia.go b/adapter/outbound/activemq/artemis_jolokia.go new file mode 100644 index 0000000..f462dfc --- /dev/null +++ b/adapter/outbound/activemq/artemis_jolokia.go @@ -0,0 +1,1144 @@ +package activemq + +import ( + "bytes" + "encoding/json" + "fmt" + "io" + "net/http" + "sort" + "strconv" + "strings" + "time" + + "github.com/xvlet/amqcli/config" + "github.com/xvlet/amqcli/domain" +) + +// ArtemisJolokiaClient implements domain.QueueRepository for Apache ActiveMQ Artemis (2.x / 3.x) +type ArtemisJolokiaClient struct { + url string + username string + password string + brokerName string + client *http.Client +} + +func NewArtemisJolokiaClient(cfg config.ActiveMQConfig) *ArtemisJolokiaClient { + url := cfg.JolokiaURL + if url == "" { + host := cfg.Host + if host == "" { + host = "127.0.0.1" + } + webPort := cfg.WebPort + if webPort == "" { + webPort = "8161" + } + // Default Artemis Jolokia path in Hawtio + url = fmt.Sprintf("http://%s:%s/console/jolokia", host, webPort) + } + + if !strings.Contains(url, "?") { + url += "?maxDepth=10&maxCollectionSize=10000&maxObjects=10000" + } + + return &ArtemisJolokiaClient{ + url: url, + username: cfg.Username, + password: cfg.Password, + brokerName: "localhost", + client: &http.Client{ + Timeout: 10 * time.Second, + }, + } +} + +// GetURL returns the active Jolokia URL +func (a *ArtemisJolokiaClient) GetURL() string { + return a.url +} + +// SetURL sets or updates the Jolokia URL +func (a *ArtemisJolokiaClient) SetURL(url string) { + if !strings.Contains(url, "?") { + url += "?maxDepth=10&maxCollectionSize=10000&maxObjects=10000" + } + a.url = url +} + +func (a *ArtemisJolokiaClient) doRequest(reqData JolokiaRequest) ([]byte, error) { + b, err := json.Marshal(reqData) + if err != nil { + return nil, err + } + + req, err := http.NewRequest(http.MethodPost, a.url, bytes.NewReader(b)) + if err != nil { + return nil, err + } + req.Header.Set("Content-Type", "application/json") + + // Set Origin header to satisfy Artemis Jolokia CORS policies (allow localhost / 127.0.0.1) + origin := "http://localhost:8161" + if parts := strings.Split(a.url, "/"); len(parts) >= 3 { + hostPort := parts[2] + scheme := parts[0] + if strings.HasPrefix(scheme, "http") { + if strings.HasPrefix(hostPort, "127.0.0.1") { + port := "" + if hp := strings.Split(hostPort, ":"); len(hp) > 1 { + port = ":" + hp[1] + } + origin = fmt.Sprintf("%s//localhost%s", scheme, port) + } else { + origin = fmt.Sprintf("%s//%s", scheme, hostPort) + } + } + } + req.Header.Set("Origin", origin) + + if a.username != "" && a.password != "" { + req.SetBasicAuth(a.username, a.password) + } + + // #nosec G704 -- internal http client using configured URL + resp, err := a.client.Do(req) + if err != nil { + return nil, err + } + defer func() { _ = resp.Body.Close() }() + + respBytes, err := io.ReadAll(resp.Body) + if err != nil { + return nil, err + } + + var baseResp struct { + Status int `json:"status"` + Error string `json:"error"` + } + if err := json.Unmarshal(respBytes, &baseResp); err == nil { + if baseResp.Status != 200 && baseResp.Status != 0 { + return nil, fmt.Errorf("artemis jolokia error (status %d): %s", baseResp.Status, baseResp.Error) + } + } else { + trimmed := strings.TrimSpace(string(respBytes)) + if strings.HasPrefix(trimmed, "<") { + if resp.StatusCode == http.StatusUnauthorized { + return nil, fmt.Errorf("jolokia authentication failed (401): check username/password") + } + if resp.StatusCode == http.StatusForbidden { + return nil, fmt.Errorf("jolokia access forbidden (403): check origins/permissions") + } + return nil, fmt.Errorf("jolokia returned HTML instead of JSON (status %d): check URL and credentials", resp.StatusCode) + } + } + + return respBytes, nil +} + +func (a *ArtemisJolokiaClient) GetBrokerStats() (domain.BrokerStats, error) { + var stats domain.BrokerStats + + // 1. Get Artemis ServerControl Stats + reqData := JolokiaRequest{ + Type: "read", + Mbean: "org.apache.activemq.artemis:broker=*", + } + + respBytes, err := a.doRequest(reqData) + if err == nil { + var result struct { + Value map[string]map[string]interface{} `json:"value"` + } + if err := json.Unmarshal(respBytes, &result); err == nil { + for mbeanKey, v := range result.Value { + // Cache brokerName from MBean key + a.extractBrokerName(mbeanKey) + + if val, ok := v["TotalMessagesAdded"].(float64); ok { + stats.TotalEnqueueCount = int64(val) + } + if val, ok := v["TotalMessagesAcknowledged"].(float64); ok { + stats.TotalDequeueCount = int64(val) + } + if val, ok := v["TotalConsumerCount"].(float64); ok { + stats.TotalConsumerCount = int64(val) + } + if val, ok := v["TotalConnectionCount"].(float64); ok { + stats.TotalProducerCount = int64(val) // Use Connection count as proxy if producer count is not global + } else if val, ok := v["ConnectionCount"].(float64); ok { + stats.TotalProducerCount = int64(val) + } + if val, ok := v["AddressMemoryUsagePercentage"].(float64); ok { + stats.MemoryPercentUsage = int(val) + } + if val, ok := v["DiskStoreUsage"].(float64); ok { + stats.StorePercentUsage = int(val) + } + if uptime, ok := v["Uptime"].(string); ok && uptime != "" { + stats.Uptime = uptime + } else if uptimeMillis, ok := v["UptimeMillis"].(float64); ok && uptimeMillis > 0 { + stats.Uptime = formatDuration(time.Duration(uptimeMillis) * time.Millisecond) + } + break + } + } + } + + // 2. Get Operating System CPU Stats + osReq := JolokiaRequest{ + Type: "read", + Mbean: "java.lang:type=OperatingSystem", + Attribute: "ProcessCpuLoad,SystemCpuLoad", + } + if osBytes, err := a.doRequest(osReq); err == nil { + var result struct { + Value struct { + ProcessCpuLoad float64 `json:"ProcessCpuLoad"` + SystemCpuLoad float64 `json:"SystemCpuLoad"` + } `json:"value"` + } + if err := json.Unmarshal(osBytes, &result); err == nil { + cpu := result.Value.ProcessCpuLoad + if cpu < 0 { + cpu = result.Value.SystemCpuLoad + } + if cpu > 0 { + stats.CPUUsage = cpu * 100.0 + } + } + } + + return stats, nil +} + +func (a *ArtemisJolokiaClient) GetBrokerInfo() (string, error) { + artemisReq := JolokiaRequest{ + Type: "read", + Mbean: "org.apache.activemq.artemis:broker=*", + Attribute: "Version", + } + + respBytes, err := a.doRequest(artemisReq) + if err != nil { + return "", err + } + + var result struct { + Value map[string]struct { + Version string `json:"Version"` + } `json:"value"` + } + if err := json.Unmarshal(respBytes, &result); err == nil && len(result.Value) > 0 { + for k, v := range result.Value { + a.extractBrokerName(k) + if v.Version != "" { + return fmt.Sprintf("Apache ActiveMQ Artemis %s", v.Version), nil + } + } + } + + var resultStr struct { + Value string `json:"value"` + } + if err := json.Unmarshal(respBytes, &resultStr); err == nil && resultStr.Value != "" { + return fmt.Sprintf("Apache ActiveMQ Artemis %s", resultStr.Value), nil + } + + return "Apache ActiveMQ Artemis", nil +} + +func (a *ArtemisJolokiaClient) GetJVMStats() (domain.JVMStats, error) { + var stats domain.JVMStats + + batchReq := []JolokiaRequest{ + { + Type: "read", + Mbean: "java.lang:type=Memory", + }, + { + Type: "read", + Mbean: "java.lang:type=Threading", + }, + } + + b, _ := json.Marshal(batchReq) + req, _ := http.NewRequest(http.MethodPost, a.url, bytes.NewReader(b)) + req.Header.Set("Content-Type", "application/json") + if a.username != "" && a.password != "" { + req.SetBasicAuth(a.username, a.password) + } + + // #nosec G704 -- internal http client + resp, err := a.client.Do(req) + if err != nil { + return stats, err + } + defer func() { _ = resp.Body.Close() }() + + if resp.StatusCode != http.StatusOK { + return stats, fmt.Errorf("unexpected status: %d", resp.StatusCode) + } + + body, err := io.ReadAll(resp.Body) + if err != nil { + return stats, err + } + + var results []struct { + Request struct { + Mbean string `json:"mbean"` + } `json:"request"` + Value map[string]interface{} `json:"value"` + } + + if err := json.Unmarshal(body, &results); err != nil { + return stats, err + } + + for _, r := range results { + switch r.Request.Mbean { + case "java.lang:type=Memory": + if heap, ok := r.Value["HeapMemoryUsage"].(map[string]interface{}); ok { + if used, ok := heap["used"].(float64); ok { + stats.HeapMemoryUsed = int64(used) + } + if max, ok := heap["max"].(float64); ok { + stats.HeapMemoryMax = int64(max) + } + } + if nonHeap, ok := r.Value["NonHeapMemoryUsage"].(map[string]interface{}); ok { + if used, ok := nonHeap["used"].(float64); ok { + stats.NonHeapMemoryUsed = int64(used) + } + } + case "java.lang:type=Threading": + if count, ok := r.Value["ThreadCount"].(float64); ok { + stats.ThreadCount = int(count) + } + if peak, ok := r.Value["PeakThreadCount"].(float64); ok { + stats.PeakThreadCount = int(peak) + } + } + } + + return stats, nil +} + +func (a *ArtemisJolokiaClient) GetQueues() ([]domain.Queue, error) { + // Query all queues across all addresses + reqData := JolokiaRequest{ + Type: "read", + Mbean: "org.apache.activemq.artemis:broker=*,component=addresses,address=*,subcomponent=queues,routing-type=*,queue=*", + } + + respBytes, err := a.doRequest(reqData) + if err != nil { + if strings.Contains(err.Error(), "status 404") || strings.Contains(err.Error(), "No MBean") { + return []domain.Queue{}, nil + } + return nil, err + } + + var result struct { + Value map[string]map[string]interface{} `json:"value"` + } + + if err := json.Unmarshal(respBytes, &result); err != nil { + return nil, err + } + + var queues []domain.Queue + for mbeanKey, props := range result.Value { + name := a.extractProperty(mbeanKey, "queue") + if name == "" { + if v, ok := props["Name"].(string); ok { + name = v + } else if v, ok := props["QueueName"].(string); ok { + name = v + } + } + + // Filter out internal multicast or temporary sub-queues if needed, but display all named queues + if name != "" { + q := domain.Queue{ + Name: name, + } + + if v, ok := props["MessageCount"].(float64); ok { + q.Pending = int64(v) + } + if v, ok := props["ConsumerCount"].(float64); ok { + q.Consumers = int64(v) + } + if v, ok := props["MessagesAdded"].(float64); ok { + q.Enqueued = int64(v) + } + if v, ok := props["MessagesAcknowledged"].(float64); ok { + q.Dequeued = int64(v) + } + if v, ok := props["DeliveringCount"].(float64); ok { + q.InFlightCount = int64(v) + } + if v, ok := props["MessagesExpired"].(float64); ok { + q.ExpiredCount = int64(v) + } + if v, ok := props["MessagesKilled"].(float64); ok && q.ExpiredCount == 0 { + q.ExpiredCount = int64(v) + } + if v, ok := props["PersistentSize"].(float64); ok { + q.StoreMessageSize = int64(v) + } + + queues = append(queues, q) + } + } + + sort.Slice(queues, func(i, j int) bool { + // DLQ to the top if present + if strings.EqualFold(queues[i].Name, "DLQ") || strings.EqualFold(queues[i].Name, "ActiveMQ.DLQ") { + return true + } + if strings.EqualFold(queues[j].Name, "DLQ") || strings.EqualFold(queues[j].Name, "ActiveMQ.DLQ") { + return false + } + return strings.ToLower(queues[i].Name) < strings.ToLower(queues[j].Name) + }) + + return queues, nil +} + +func (a *ArtemisJolokiaClient) GetTopics() ([]domain.Topic, error) { + // In Artemis, topics are multicast queues or addresses with multicast routing + reqData := JolokiaRequest{ + Type: "read", + Mbean: "org.apache.activemq.artemis:broker=*,component=addresses,address=*,subcomponent=queues,routing-type=\"multicast\",queue=*", + } + + respBytes, err := a.doRequest(reqData) + if err != nil { + if strings.Contains(err.Error(), "status 404") || strings.Contains(err.Error(), "No MBean") { + return []domain.Topic{}, nil + } + return nil, err + } + + var result struct { + Value map[string]map[string]interface{} `json:"value"` + } + + if err := json.Unmarshal(respBytes, &result); err != nil { + return nil, err + } + + var topics []domain.Topic + for mbeanKey, props := range result.Value { + name := a.extractProperty(mbeanKey, "address") + if name == "" { + name = a.extractProperty(mbeanKey, "queue") + } + + if name != "" { + t := domain.Topic{ + Name: name, + } + if v, ok := props["MessagesAdded"].(float64); ok { + t.EnqueueCount = int64(v) + } + if v, ok := props["MessagesAcknowledged"].(float64); ok { + t.DequeueCount = int64(v) + } + if v, ok := props["ConsumerCount"].(float64); ok { + t.ConsumerCount = int64(v) + } + topics = append(topics, t) + } + } + + sort.Slice(topics, func(i, j int) bool { + return strings.ToLower(topics[i].Name) < strings.ToLower(topics[j].Name) + }) + + return topics, nil +} + +func (a *ArtemisJolokiaClient) GetConnections() ([]domain.Connection, error) { + // 1. Try listConnectionsAsJSON() operation on ActiveMQServerControl + reqData := JolokiaRequest{ + Type: "exec", + Mbean: a.getServerControlMBean(), + Operation: "listConnectionsAsJSON()", + } + + respBytes, err := a.doRequest(reqData) + if err == nil { + var rawJSON string + var resultStr struct { + Value string `json:"value"` + } + if err := json.Unmarshal(respBytes, &resultStr); err == nil && resultStr.Value != "" { + rawJSON = resultStr.Value + } else { + var resultMap struct { + Value map[string]string `json:"value"` + } + if err := json.Unmarshal(respBytes, &resultMap); err == nil && len(resultMap.Value) > 0 { + for k, v := range resultMap.Value { + a.extractBrokerName(k) + rawJSON = v + break + } + } + } + + if rawJSON != "" { + var connList []struct { + ConnectionID string `json:"connectionID"` + ClientAddress string `json:"clientAddress"` + RemoteAddress string `json:"remoteAddress"` + CreationTime int64 `json:"creationTime"` + } + if err := json.Unmarshal([]byte(rawJSON), &connList); err == nil && len(connList) > 0 { + var connections []domain.Connection + for _, c := range connList { + addr := c.ClientAddress + if addr == "" { + addr = c.RemoteAddress + } + addr = strings.TrimPrefix(addr, "/") + addr = strings.TrimPrefix(addr, "tcp://") + + connections = append(connections, domain.Connection{ + Name: c.ConnectionID, + RemoteAddress: addr, + Active: true, + Slow: false, + }) + } + sort.Slice(connections, func(i, j int) bool { + return strings.ToLower(connections[i].Name) < strings.ToLower(connections[j].Name) + }) + return connections, nil + } + } + } + + // 2. Fallback: Query connection MBeans directly + connReq := JolokiaRequest{ + Type: "read", + Mbean: "org.apache.activemq.artemis:broker=*,component=connections,connection=*", + } + if connBytes, err := a.doRequest(connReq); err == nil { + var connRes struct { + Value map[string]map[string]interface{} `json:"value"` + } + if err := json.Unmarshal(connBytes, &connRes); err == nil { + var connections []domain.Connection + for k, v := range connRes.Value { + name := a.extractProperty(k, "connection") + addr := "" + if aVal, ok := v["RemoteAddress"].(string); ok { + addr = aVal + } else if aVal, ok := v["ClientAddress"].(string); ok { + addr = aVal + } + addr = strings.TrimPrefix(addr, "/") + addr = strings.TrimPrefix(addr, "tcp://") + + if name != "" { + connections = append(connections, domain.Connection{ + Name: name, + RemoteAddress: addr, + Active: true, + }) + } + } + sort.Slice(connections, func(i, j int) bool { + return strings.ToLower(connections[i].Name) < strings.ToLower(connections[j].Name) + }) + return connections, nil + } + } + + return []domain.Connection{}, nil +} + +func (a *ArtemisJolokiaClient) GetAllConsumers() ([]domain.Consumer, error) { + // 1. Try listConsumersAsJSON(java.lang.String) operation on ActiveMQServerControl + reqData := JolokiaRequest{ + Type: "exec", + Mbean: a.getServerControlMBean(), + Operation: "listConsumersAsJSON(java.lang.String)", + Arguments: []interface{}{"{}"}, + } + + respBytes, err := a.doRequest(reqData) + if err == nil { + var rawJSON string + var resultStr struct { + Value string `json:"value"` + } + if err := json.Unmarshal(respBytes, &resultStr); err == nil && resultStr.Value != "" { + rawJSON = resultStr.Value + } else { + var resultMap struct { + Value map[string]string `json:"value"` + } + if err := json.Unmarshal(respBytes, &resultMap); err == nil && len(resultMap.Value) > 0 { + for k, v := range resultMap.Value { + a.extractBrokerName(k) + rawJSON = v + break + } + } + } + + if rawJSON != "" { + var rawConsumers []struct { + ConsumerID interface{} `json:"consumerID"` + ConnectionID string `json:"connectionID"` + SessionID string `json:"sessionID"` + QueueName string `json:"queueName"` + Address string `json:"address"` + RemoteAddress string `json:"remoteAddress"` + ClientAddress string `json:"clientAddress"` + DeliveringCount int64 `json:"deliveringCount"` + MessagesAcknowledged int64 `json:"messagesAcknowledged"` + CreationTime int64 `json:"creationTime"` + } + if err := json.Unmarshal([]byte(rawJSON), &rawConsumers); err == nil { + var consumers []domain.Consumer + for _, rc := range rawConsumers { + addr := rc.RemoteAddress + if addr == "" { + addr = rc.ClientAddress + } + addr = strings.TrimPrefix(addr, "/") + addr = strings.TrimPrefix(addr, "tcp://") + + dest := rc.QueueName + if dest == "" { + dest = rc.Address + } + + cID := fmt.Sprintf("%v", rc.ConsumerID) + c := domain.Consumer{ + ConsumerID: cID, + ConnectionID: rc.ConnectionID, + ClientID: rc.SessionID, + DestinationName: dest, + RemoteAddress: addr, + Dequeues: rc.MessagesAcknowledged, + PendingQueueSize: rc.DeliveringCount, + } + + pid, uptime := a.parseConsumerInfo(c.ClientID, c.ConnectionID) + if rc.CreationTime > 0 { + dur := time.Since(time.UnixMilli(rc.CreationTime)) + if dur > 0 { + uptime = formatDuration(dur) + } + } + c.PID = pid + c.Uptime = uptime + + consumers = append(consumers, c) + } + + sort.Slice(consumers, func(i, j int) bool { + return strings.ToLower(consumers[i].DestinationName) < strings.ToLower(consumers[j].DestinationName) + }) + return consumers, nil + } + } + } + + // 2. Fallback: Query consumer MBeans directly + mbeanPattern := "org.apache.activemq.artemis:broker=*,component=addresses,address=*,subcomponent=queues,routing-type=*,queue=*,subcomponent=consumers,consumer-id=*" + consReq := JolokiaRequest{ + Type: "read", + Mbean: mbeanPattern, + } + + consBytes, err := a.doRequest(consReq) + if err != nil { + return []domain.Consumer{}, nil + } + + var mbeanResult struct { + Value map[string]map[string]interface{} `json:"value"` + } + if err := json.Unmarshal(consBytes, &mbeanResult); err != nil { + return []domain.Consumer{}, nil + } + + var consumers []domain.Consumer + for k, props := range mbeanResult.Value { + cID := a.extractProperty(k, "consumer-id") + qName := a.extractProperty(k, "queue") + + c := domain.Consumer{ + ConsumerID: cID, + DestinationName: qName, + } + + if v, ok := props["ConnectionID"].(string); ok { + c.ConnectionID = v + } + if v, ok := props["SessionID"].(string); ok { + c.ClientID = v + } + if v, ok := props["RemoteAddress"].(string); ok { + c.RemoteAddress = strings.TrimPrefix(strings.TrimPrefix(v, "/"), "tcp://") + } + if v, ok := props["MessagesAcknowledged"].(float64); ok { + c.Dequeues = int64(v) + } + if v, ok := props["DeliveringCount"].(float64); ok { + c.PendingQueueSize = int64(v) + } + + pid, uptime := a.parseConsumerInfo(c.ClientID, c.ConnectionID) + c.PID = pid + c.Uptime = uptime + + consumers = append(consumers, c) + } + + sort.Slice(consumers, func(i, j int) bool { + return strings.ToLower(consumers[i].DestinationName) < strings.ToLower(consumers[j].DestinationName) + }) + + return consumers, nil +} + +func (a *ArtemisJolokiaClient) GetQueueDetail(name string) (*domain.QueueDetail, error) { + // Query specific queue MBean using wildcard address to support any address binding + mbeanPattern := fmt.Sprintf("org.apache.activemq.artemis:broker=*,component=addresses,address=*,subcomponent=queues,routing-type=*,queue=%s", name) + if !strings.Contains(name, "\"") { + mbeanPattern = fmt.Sprintf("org.apache.activemq.artemis:broker=*,component=addresses,address=*,subcomponent=queues,routing-type=*,queue=\"%s\"", name) + } + + reqData := JolokiaRequest{ + Type: "read", + Mbean: mbeanPattern, + } + + respBytes, err := a.doRequest(reqData) + if err != nil { + // Fallback without quotes + mbeanPattern = fmt.Sprintf("org.apache.activemq.artemis:broker=*,component=addresses,address=*,subcomponent=queues,routing-type=*,queue=%s", name) + reqData.Mbean = mbeanPattern + respBytes, err = a.doRequest(reqData) + if err != nil { + return nil, err + } + } + + var result struct { + Value map[string]map[string]interface{} `json:"value"` + } + if err := json.Unmarshal(respBytes, &result); err != nil { + return nil, err + } + + qd := &domain.QueueDetail{ + Name: name, + } + + var specificMBean string + for mbeanKey, props := range result.Value { + specificMBean = mbeanKey + if v, ok := props["MessageCount"].(float64); ok { + qd.QueueSize = int64(v) + } + if v, ok := props["ConsumerCount"].(float64); ok { + qd.ConsumerCount = int64(v) + } + if v, ok := props["MessagesAdded"].(float64); ok { + qd.EnqueueCount = int64(v) + } + if v, ok := props["MessagesAcknowledged"].(float64); ok { + qd.DequeueCount = int64(v) + } + if v, ok := props["DeliveringCount"].(float64); ok { + qd.InFlightCount = int64(v) + } + if v, ok := props["MessagesExpired"].(float64); ok { + qd.ExpiredCount = int64(v) + } + if v, ok := props["MessagesKilled"].(float64); ok && qd.ExpiredCount == 0 { + qd.ExpiredCount = int64(v) + } + if v, ok := props["PersistentSize"].(float64); ok { + qd.StoreMessageSize = int64(v) + } + break + } + + // Resolve connection RemoteAddress map + connMap := make(map[string]string) + conns, _ := a.GetConnections() + for _, conn := range conns { + connMap[conn.Name] = conn.RemoteAddress + } + + // Directly query consumers on this specific queue MBean + if specificMBean != "" { + consReq := JolokiaRequest{ + Type: "exec", + Mbean: specificMBean, + Operation: "listConsumersAsJSON()", + } + if cBytes, err := a.doRequest(consReq); err == nil { + var cResult struct { + Value string `json:"value"` + } + if err := json.Unmarshal(cBytes, &cResult); err == nil && cResult.Value != "" { + var rawCons []struct { + ConsumerID interface{} `json:"consumerID"` + SequentialID interface{} `json:"sequentialId"` + ConnectionID string `json:"connectionID"` + SessionID string `json:"sessionID"` + MessagesAcknowledged int64 `json:"messagesAcknowledged"` + CreationTime int64 `json:"creationTime"` + } + if err := json.Unmarshal([]byte(cResult.Value), &rawCons); err == nil { + for _, rc := range rawCons { + cID := fmt.Sprintf("%v", rc.ConsumerID) + remoteAddr := connMap[rc.ConnectionID] + if remoteAddr == "" { + remoteAddr = "127.0.0.1" + } + + uptime := "-" + if rc.CreationTime > 0 { + dur := time.Since(time.UnixMilli(rc.CreationTime)) + if dur > 0 { + uptime = formatDuration(dur) + } + } + + pid, _ := a.parseConsumerInfo(rc.SessionID, rc.ConnectionID) + qd.Consumers = append(qd.Consumers, domain.Consumer{ + ConsumerID: cID, + ConnectionID: rc.ConnectionID, + ClientID: rc.SessionID, + DestinationName: name, + RemoteAddress: remoteAddr, + Dequeues: rc.MessagesAcknowledged, + PID: pid, + Uptime: uptime, + }) + } + } + } + } + } + + // Fallback to GetAllConsumers if specific MBean query returned nothing but ConsumerCount > 0 + if len(qd.Consumers) == 0 && qd.ConsumerCount > 0 { + allConsumers, err := a.GetAllConsumers() + if err == nil { + for _, c := range allConsumers { + if strings.EqualFold(c.DestinationName, name) { + qd.Consumers = append(qd.Consumers, c) + } + } + } + } + + return qd, nil +} + +func (a *ArtemisJolokiaClient) CreateQueue(name string) error { + // In Artemis, createQueue on ActiveMQServerControl takes (address, queueName, routingType) + // e.g. createQueue(java.lang.String,java.lang.String,java.lang.String) + reqData := JolokiaRequest{ + Type: "exec", + Mbean: a.getServerControlMBean(), + Operation: "createQueue(java.lang.String,java.lang.String,java.lang.String)", + Arguments: []interface{}{name, name, "ANYCAST"}, + } + + _, err := a.doRequest(reqData) + if err != nil { + // Fallback for older Artemis versions: createAddress or 2-arg createQueue + reqData.Operation = "createQueue(java.lang.String,java.lang.String)" + reqData.Arguments = []interface{}{name, name} + _, err2 := a.doRequest(reqData) + if err2 == nil { + return nil + } + return err + } + return nil +} + +func (a *ArtemisJolokiaClient) DeleteQueue(name string) error { + // In Artemis: destroyQueue(java.lang.String,boolean) or destroyQueue(java.lang.String) + reqData := JolokiaRequest{ + Type: "exec", + Mbean: a.getServerControlMBean(), + Operation: "destroyQueue(java.lang.String,boolean)", + Arguments: []interface{}{name, true}, + } + + _, err := a.doRequest(reqData) + if err != nil { + // Fallback for single arg destroyQueue + reqData.Operation = "destroyQueue(java.lang.String)" + reqData.Arguments = []interface{}{name} + _, err2 := a.doRequest(reqData) + if err2 == nil { + return nil + } + return err + } + return nil +} + +func (a *ArtemisJolokiaClient) PurgeQueue(name string) error { + queueMBean, err := a.findQueueMBean(name) + if err != nil { + return err + } + + reqData := JolokiaRequest{ + Type: "exec", + Mbean: queueMBean, + Operation: "removeAllMessages()", + } + + _, err = a.doRequest(reqData) + return err +} + +func (a *ArtemisJolokiaClient) RemoveMessage(queueName string, messageID string) error { + queueMBean, err := a.findQueueMBean(queueName) + if err != nil { + return err + } + + // Artemis removeMessage can take long messageID or string filter + if numID, err := strconv.ParseInt(messageID, 10, 64); err == nil { + reqData := JolokiaRequest{ + Type: "exec", + Mbean: queueMBean, + Operation: "removeMessage(long)", + Arguments: []interface{}{numID}, + } + if _, err := a.doRequest(reqData); err == nil { + return nil + } + } + + // Fallback using filter + filter := fmt.Sprintf("JMSMessageID = '%s' OR AMQMessageID = '%s'", messageID, messageID) + reqData := JolokiaRequest{ + Type: "exec", + Mbean: queueMBean, + Operation: "removeMessages(java.lang.String)", + Arguments: []interface{}{filter}, + } + _, err = a.doRequest(reqData) + return err +} + +func (a *ArtemisJolokiaClient) MoveMessage(queueName string, messageID string, destQueue string) error { + queueMBean, err := a.findQueueMBean(queueName) + if err != nil { + return err + } + + filter := fmt.Sprintf("JMSMessageID = '%s' OR AMQMessageID = '%s'", messageID, messageID) + if _, err := strconv.ParseInt(messageID, 10, 64); err == nil { + filter = fmt.Sprintf("AMQMessageID = %s OR JMSMessageID = '%s'", messageID, messageID) + } + + reqData := JolokiaRequest{ + Type: "exec", + Mbean: queueMBean, + Operation: "moveMessages(java.lang.String,java.lang.String)", + Arguments: []interface{}{filter, destQueue}, + } + + _, err = a.doRequest(reqData) + return err +} + +func (a *ArtemisJolokiaClient) CopyMessage(queueName string, messageID string, destQueue string) error { + queueMBean, err := a.findQueueMBean(queueName) + if err != nil { + return err + } + + filter := fmt.Sprintf("JMSMessageID = '%s' OR AMQMessageID = '%s'", messageID, messageID) + if _, err := strconv.ParseInt(messageID, 10, 64); err == nil { + filter = fmt.Sprintf("AMQMessageID = %s OR JMSMessageID = '%s'", messageID, messageID) + } + + reqData := JolokiaRequest{ + Type: "exec", + Mbean: queueMBean, + Operation: "copyMessages(java.lang.String,java.lang.String)", + Arguments: []interface{}{filter, destQueue}, + } + + _, err = a.doRequest(reqData) + return err +} + +func (a *ArtemisJolokiaClient) RetryMessage(queueName string, messageID string) error { + queueMBean, err := a.findQueueMBean(queueName) + if err != nil { + return err + } + + if numID, err := strconv.ParseInt(messageID, 10, 64); err == nil { + reqData := JolokiaRequest{ + Type: "exec", + Mbean: queueMBean, + Operation: "retryMessage(long)", + Arguments: []interface{}{numID}, + } + if _, err := a.doRequest(reqData); err == nil { + return nil + } + } + + filter := fmt.Sprintf("JMSMessageID = '%s' OR AMQMessageID = '%s'", messageID, messageID) + reqData := JolokiaRequest{ + Type: "exec", + Mbean: queueMBean, + Operation: "retryMessages(java.lang.String)", + Arguments: []interface{}{filter}, + } + + _, err = a.doRequest(reqData) + return err +} + +// Helper methods + +func (a *ArtemisJolokiaClient) getServerControlMBean() string { + if a.brokerName != "" && a.brokerName != "localhost" { + return fmt.Sprintf("org.apache.activemq.artemis:broker=\"%s\"", a.brokerName) + } + return "org.apache.activemq.artemis:broker=*" +} + +func (a *ArtemisJolokiaClient) findQueueMBean(queueName string) (string, error) { + // Query for queue MBean name + pattern := fmt.Sprintf("org.apache.activemq.artemis:broker=*,component=addresses,address=*,subcomponent=queues,routing-type=*,queue=\"%s\"", queueName) + reqData := JolokiaRequest{ + Type: "read", + Mbean: pattern, + } + + respBytes, err := a.doRequest(reqData) + if err != nil || len(respBytes) == 0 { + // Fallback without quotes + pattern = fmt.Sprintf("org.apache.activemq.artemis:broker=*,component=addresses,address=*,subcomponent=queues,routing-type=*,queue=%s", queueName) + reqData.Mbean = pattern + respBytes, err = a.doRequest(reqData) + if err != nil { + return "", fmt.Errorf("queue mbean not found for '%s': %w", queueName, err) + } + } + + var result struct { + Value map[string]interface{} `json:"value"` + } + if err := json.Unmarshal(respBytes, &result); err == nil && len(result.Value) > 0 { + for mbean := range result.Value { + return mbean, nil + } + } + + // Default template if query returned empty + if a.brokerName != "" && a.brokerName != "localhost" { + return fmt.Sprintf("org.apache.activemq.artemis:broker=\"%s\",component=addresses,address=\"%s\",subcomponent=queues,routing-type=\"anycast\",queue=\"%s\"", a.brokerName, queueName, queueName), nil + } + return fmt.Sprintf("org.apache.activemq.artemis:broker=*,component=addresses,address=\"%s\",subcomponent=queues,routing-type=\"anycast\",queue=\"%s\"", queueName, queueName), nil +} + +func (a *ArtemisJolokiaClient) extractProperty(mbeanStr, propName string) string { + parts := strings.Split(mbeanStr, ",") + for _, p := range parts { + kv := strings.SplitN(p, "=", 2) + if len(kv) == 2 && strings.TrimSpace(kv[0]) == propName { + val := strings.TrimSpace(kv[1]) + val = strings.Trim(val, "\"") + return val + } + } + return "" +} + +func (a *ArtemisJolokiaClient) extractBrokerName(mbeanStr string) { + if a.brokerName == "localhost" || a.brokerName == "" { + bn := a.extractProperty(mbeanStr, "broker") + if bn != "" { + a.brokerName = bn + } + } +} + +func (a *ArtemisJolokiaClient) parseConsumerInfo(clientID, connectionID string) (string, string) { + isDefaultCID := strings.HasPrefix(clientID, "ID:") + isDefaultConnID := strings.HasPrefix(connectionID, "ID:") + + cleanCID := strings.ReplaceAll(clientID, ":", "-") + cleanCID = strings.ReplaceAll(cleanCID, "_", "-") + cleanConnID := strings.ReplaceAll(connectionID, ":", "-") + cleanConnID = strings.ReplaceAll(cleanConnID, "_", "-") + + pid := "-" + uptime := "-" + + if !isDefaultCID { + if match := pidRegex.FindStringSubmatch(cleanCID); len(match) > 1 { + pid = match[1] + } + } + if pid == "-" && !isDefaultConnID { + if match := pidRegex.FindStringSubmatch(cleanConnID); len(match) > 1 { + pid = match[1] + } + } + + var ts int64 + foundTS := false + if match := timestampRegex.FindStringSubmatch(cleanCID); len(match) > 1 { + if val, err := strconv.ParseInt(match[1], 10, 64); err == nil { + ts = val + foundTS = true + } + } + + if foundTS { + var connectedAt time.Time + strTS := strconv.FormatInt(ts, 10) + switch len(strTS) { + case 10: + connectedAt = time.Unix(ts, 0) + case 13: + connectedAt = time.Unix(ts/1000, (ts%1000)*1e6) + case 14: + if t, err := time.Parse("20060102150405", strTS); err == nil { + connectedAt = t + } + } + if !connectedAt.IsZero() { + duration := time.Since(connectedAt) + if duration > 0 { + uptime = formatDuration(duration) + } + } + } + + return pid, uptime +} diff --git a/adapter/outbound/activemq/artemis_jolokia_test.go b/adapter/outbound/activemq/artemis_jolokia_test.go new file mode 100644 index 0000000..ef9e09b --- /dev/null +++ b/adapter/outbound/activemq/artemis_jolokia_test.go @@ -0,0 +1,259 @@ +package activemq + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/xvlet/amqcli/config" +) + +func TestArtemisJolokiaClient_AllOperations(t *testing.T) { + // Mock Jolokia Server for Artemis + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var req JolokiaRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + // Check if batch request + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(`[ + {"request":{"mbean":"java.lang:type=Memory"},"value":{"HeapMemoryUsage":{"used":1048576,"max":2097152},"NonHeapMemoryUsage":{"used":524288}}}, + {"request":{"mbean":"java.lang:type=Threading"},"value":{"ThreadCount":42,"PeakThreadCount":50}} + ]`)) + return + } + + w.Header().Set("Content-Type", "application/json") + + // 1. Broker Info / Version + if req.Type == "read" && req.Mbean == "org.apache.activemq.artemis:broker=*" && req.Attribute == "Version" { + _, _ = w.Write([]byte(`{ + "status": 200, + "value": { + "org.apache.activemq.artemis:broker=\"artemis-broker\"": { + "Version": "2.33.0" + } + } + }`)) + return + } + + // 2. Broker Stats + if req.Type == "read" && req.Mbean == "org.apache.activemq.artemis:broker=*" { + _, _ = w.Write([]byte(`{ + "status": 200, + "value": { + "org.apache.activemq.artemis:broker=\"artemis-broker\"": { + "TotalMessagesAdded": 1500, + "TotalMessagesAcknowledged": 1200, + "TotalConsumerCount": 8, + "TotalConnectionCount": 5, + "AddressMemoryUsagePercentage": 25, + "DiskStoreUsage": 10, + "Uptime": "2h 30m" + } + } + }`)) + return + } + + // 3. OperatingSystem CPU + if req.Type == "read" && req.Mbean == "java.lang:type=OperatingSystem" { + _, _ = w.Write([]byte(`{ + "status": 200, + "value": { + "ProcessCpuLoad": 0.15, + "SystemCpuLoad": 0.25 + } + }`)) + return + } + + // 4. Queues List + if req.Type == "read" && strings.Contains(req.Mbean, "subcomponent=queues") && !strings.Contains(req.Mbean, "multicast") { + _, _ = w.Write([]byte(`{ + "status": 200, + "value": { + "org.apache.activemq.artemis:broker=\"artemis-broker\",component=addresses,address=\"ORDER.QUEUE\",queue=\"ORDER.QUEUE\",routing-type=\"anycast\",subcomponent=queues": { + "MessageCount": 50, + "ConsumerCount": 2, + "MessagesAdded": 500, + "MessagesAcknowledged": 450, + "DeliveringCount": 3, + "MessagesExpired": 1, + "PersistentSize": 20480 + }, + "org.apache.activemq.artemis:broker=\"artemis-broker\",component=addresses,address=\"DLQ\",queue=\"DLQ\",routing-type=\"anycast\",subcomponent=queues": { + "MessageCount": 5, + "ConsumerCount": 0, + "MessagesAdded": 5, + "MessagesAcknowledged": 0, + "DeliveringCount": 0, + "MessagesExpired": 0, + "PersistentSize": 1024 + } + } + }`)) + return + } + + // 5. Topics List (Multicast) + if req.Type == "read" && strings.Contains(req.Mbean, "multicast") { + _, _ = w.Write([]byte(`{ + "status": 200, + "value": { + "org.apache.activemq.artemis:broker=\"artemis-broker\",component=addresses,address=\"EVENTS.TOPIC\",queue=\"EVENTS.TOPIC\",routing-type=\"multicast\",subcomponent=queues": { + "MessagesAdded": 100, + "MessagesAcknowledged": 100, + "ConsumerCount": 5 + } + } + }`)) + return + } + + // 6. Connections List (listConnectionsAsJSON) + if req.Type == "exec" && req.Operation == "listConnectionsAsJSON()" { + _, _ = w.Write([]byte(`{ + "status": 200, + "value": "[{\"connectionID\":\"conn-1\",\"clientAddress\":\"/127.0.0.1:54321\",\"creationTime\":1690000000000}]" + }`)) + return + } + + // 7. Consumers List (listConsumersAsJSON) + if req.Type == "exec" && (req.Operation == "listConsumersAsJSON()" || req.Operation == "listConsumersAsJSON(java.lang.String)") { + _, _ = w.Write([]byte(`{ + "status": 200, + "value": "[{\"consumerID\":\"cons-1\",\"connectionID\":\"conn-1\",\"sessionID\":\"sess-12345-1690000000000\",\"queueName\":\"ORDER.QUEUE\",\"remoteAddress\":\"/127.0.0.1:54321\",\"deliveringCount\":1,\"messagesAcknowledged\":20,\"creationTime\":1690000000000}]" + }`)) + return + } + + // 8. Exec operations (createQueue, destroyQueue, removeAllMessages, etc.) + if req.Type == "exec" { + _, _ = w.Write([]byte(`{"status": 200, "value": true}`)) + return + } + + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(`{"status": 200, "value": {}}`)) + })) + defer server.Close() + + cfg := config.ActiveMQConfig{ + Host: "127.0.0.1", + BrokerType: "artemis", + JolokiaURL: server.URL, + } + + client := NewArtemisJolokiaClient(cfg) + + // Test 1: GetBrokerInfo + info, err := client.GetBrokerInfo() + if err != nil || !strings.Contains(info, "2.33.0") { + t.Fatalf("GetBrokerInfo failed: %v, info: %s", err, info) + } + + // Test 2: GetBrokerStats + stats, err := client.GetBrokerStats() + if err != nil || stats.TotalEnqueueCount != 1500 || stats.TotalDequeueCount != 1200 { + t.Fatalf("GetBrokerStats failed: %v, stats: %+v", err, stats) + } + + // Test 3: GetJVMStats + jvmStats, err := client.GetJVMStats() + if err != nil || jvmStats.ThreadCount != 42 { + t.Fatalf("GetJVMStats failed: %v, stats: %+v", err, jvmStats) + } + + // Test 4: GetQueues + queues, err := client.GetQueues() + if err != nil || len(queues) != 2 { + t.Fatalf("GetQueues failed: %v, queues: %+v", err, queues) + } + // DLQ should be sorted first + if queues[0].Name != "DLQ" { + t.Errorf("Expected DLQ to be sorted first, got %s", queues[0].Name) + } + if queues[1].Pending != 50 || queues[1].Enqueued != 500 { + t.Errorf("Expected ORDER.QUEUE pending=50, enqueued=500, got: %+v", queues[1]) + } + + // Test 5: GetTopics + topics, err := client.GetTopics() + if err != nil || len(topics) != 1 || topics[0].Name != "EVENTS.TOPIC" { + t.Fatalf("GetTopics failed: %v, topics: %+v", err, topics) + } + + // Test 6: GetConnections + connections, err := client.GetConnections() + if err != nil || len(connections) != 1 || connections[0].Name != "conn-1" { + t.Fatalf("GetConnections failed: %v, connections: %+v", err, connections) + } + + // Test 7: GetAllConsumers + consumers, err := client.GetAllConsumers() + if err != nil || len(consumers) != 1 || consumers[0].ConsumerID != "cons-1" { + t.Fatalf("GetAllConsumers failed: %v, consumers: %+v", err, consumers) + } + + // Test 8: GetQueueDetail + qd, err := client.GetQueueDetail("ORDER.QUEUE") + if err != nil || qd.Name != "ORDER.QUEUE" || len(qd.Consumers) != 1 { + t.Fatalf("GetQueueDetail failed: %v, qd: %+v", err, qd) + } + + // Test 9: Management Operations + if err := client.CreateQueue("NEW.QUEUE"); err != nil { + t.Errorf("CreateQueue failed: %v", err) + } + if err := client.DeleteQueue("OLD.QUEUE"); err != nil { + t.Errorf("DeleteQueue failed: %v", err) + } + if err := client.PurgeQueue("ORDER.QUEUE"); err != nil { + t.Errorf("PurgeQueue failed: %v", err) + } + if err := client.RemoveMessage("ORDER.QUEUE", "12345"); err != nil { + t.Errorf("RemoveMessage failed: %v", err) + } + if err := client.MoveMessage("ORDER.QUEUE", "12345", "OTHER.QUEUE"); err != nil { + t.Errorf("MoveMessage failed: %v", err) + } + if err := client.CopyMessage("ORDER.QUEUE", "12345", "TEMP.QUEUE"); err != nil { + t.Errorf("CopyMessage failed: %v", err) + } + if err := client.RetryMessage("DLQ", "12345"); err != nil { + t.Errorf("RetryMessage failed: %v", err) + } +} + +func TestFactory_DetectQueueRepository(t *testing.T) { + // Mock Server that responds to Artemis probe + artemisServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var req JolokiaRequest + _ = json.NewDecoder(r.Body).Decode(&req) + if req.Mbean == "org.apache.activemq.artemis:broker=*" { + _, _ = w.Write([]byte(`{"status": 200, "value": {"org.apache.activemq.artemis:broker=\"artemis\"": {"Version": "2.31.0"}}}`)) + return + } + w.WriteHeader(http.StatusNotFound) + _, _ = w.Write([]byte(`{"status": 404, "error": "No MBean"}`)) + })) + defer artemisServer.Close() + + cfg := config.ActiveMQConfig{ + Host: "127.0.0.1", + BrokerType: "auto", + JolokiaURL: artemisServer.URL, + } + + repo, detected := DetectQueueRepository(cfg) + if detected != "artemis" { + t.Errorf("Expected detected broker to be artemis, got %s", detected) + } + if _, ok := repo.(*ArtemisJolokiaClient); !ok { + t.Errorf("Expected *ArtemisJolokiaClient type, got %T", repo) + } +} diff --git a/adapter/outbound/activemq/factory.go b/adapter/outbound/activemq/factory.go new file mode 100644 index 0000000..b744e25 --- /dev/null +++ b/adapter/outbound/activemq/factory.go @@ -0,0 +1,86 @@ +package activemq + +import ( + "fmt" + "net/http" + "strings" + "time" + + "github.com/xvlet/amqcli/config" + "github.com/xvlet/amqcli/domain" +) + +// NewQueueRepository creates an appropriate domain.QueueRepository based on configuration or auto-detection. +func NewQueueRepository(cfg config.ActiveMQConfig) domain.QueueRepository { + if strings.EqualFold(cfg.BrokerType, "artemis") { + return NewArtemisJolokiaClient(cfg) + } + if strings.EqualFold(cfg.BrokerType, "classic") { + return NewJolokiaClient(cfg) + } + + // Auto-detection + repo, _ := DetectQueueRepository(cfg) + return repo +} + +// DetectQueueRepository probes target endpoints to automatically detect broker type (Classic vs Artemis) +func DetectQueueRepository(cfg config.ActiveMQConfig) (domain.QueueRepository, string) { + var candidateURLs []string + if cfg.JolokiaURL != "" { + candidateURLs = append(candidateURLs, cfg.JolokiaURL) + } + + webPort := cfg.WebPort + if webPort == "" { + webPort = "8161" + } + + standardPaths := []string{ + fmt.Sprintf("http://%s:%s/console/jolokia", cfg.Host, webPort), + fmt.Sprintf("http://%s:%s/api/jolokia", cfg.Host, webPort), + fmt.Sprintf("http://%s:%s/jolokia", cfg.Host, webPort), + } + + for _, p := range standardPaths { + found := false + for _, u := range candidateURLs { + if strings.EqualFold(strings.Split(u, "?")[0], p) { + found = true + break + } + } + if !found { + candidateURLs = append(candidateURLs, p) + } + } + + probeClient := &http.Client{ + Timeout: 2 * time.Second, + } + + for _, rawURL := range candidateURLs { + // 1. Probe Artemis + artemisCfg := cfg + artemisCfg.JolokiaURL = rawURL + artemisClient := NewArtemisJolokiaClient(artemisCfg) + artemisClient.client = probeClient + if info, err := artemisClient.GetBrokerInfo(); err == nil && info != "" { + artemisClient.client = &http.Client{Timeout: 10 * time.Second} + return artemisClient, "artemis" + } + + // 2. Probe Classic + classicCfg := cfg + classicCfg.JolokiaURL = rawURL + classicClient := NewJolokiaClient(classicCfg) + classicClient.client = probeClient + if info, err := classicClient.GetBrokerInfo(); err == nil && info != "" { + classicClient.client = &http.Client{Timeout: 10 * time.Second} + return classicClient, "classic" + } + } + + // Default fallback to Classic + return NewJolokiaClient(cfg), "classic" +} diff --git a/adapter/outbound/activemq/jolokia.go b/adapter/outbound/activemq/jolokia.go index c3395cf..de18171 100644 --- a/adapter/outbound/activemq/jolokia.go +++ b/adapter/outbound/activemq/jolokia.go @@ -30,8 +30,19 @@ type JolokiaClient struct { } func NewJolokiaClient(cfg config.ActiveMQConfig) *JolokiaClient { - // Ensure Jolokia URL has parameters to prevent truncation url := cfg.JolokiaURL + if url == "" { + host := cfg.Host + if host == "" { + host = "127.0.0.1" + } + webPort := cfg.WebPort + if webPort == "" { + webPort = "8161" + } + url = fmt.Sprintf("http://%s:%s/api/jolokia", host, webPort) + } + if !strings.Contains(url, "?") { url += "?maxDepth=10&maxCollectionSize=10000&maxObjects=10000" } @@ -69,13 +80,24 @@ func (j *JolokiaClient) doRequest(reqData JolokiaRequest) ([]byte, error) { } req.Header.Set("Content-Type", "application/json") - // Add Origin header to bypass Jolokia CORS strict checking - req.Header.Set("Origin", "http://localhost") - // Or parse from url, but ActiveMQ usually just needs it to not be absent/null for strict setups unless specifically configured. - // Actually, just passing the Jolokia URL's domain/scheme works. - if parts := strings.Split(j.url, "/api/jolokia"); len(parts) > 0 { - req.Header.Set("Origin", parts[0]) + // Set Origin header to satisfy Jolokia CORS policies (allow localhost / 127.0.0.1) + origin := "http://localhost:8161" + if parts := strings.Split(j.url, "/"); len(parts) >= 3 { + hostPort := parts[2] + scheme := parts[0] + if strings.HasPrefix(scheme, "http") { + if strings.HasPrefix(hostPort, "127.0.0.1") { + port := "" + if hp := strings.Split(hostPort, ":"); len(hp) > 1 { + port = ":" + hp[1] + } + origin = fmt.Sprintf("%s//localhost%s", scheme, port) + } else { + origin = fmt.Sprintf("%s//%s", scheme, hostPort) + } + } } + req.Header.Set("Origin", origin) if j.username != "" && j.password != "" { req.SetBasicAuth(j.username, j.password) diff --git a/cmd/main.go b/cmd/main.go index ec4bba6..3b18396 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -42,7 +42,7 @@ func main() { } // 2. Initialize outbound adapters - jolokiaClient := activemq.NewJolokiaClient(mqConfig) + queueRepo := activemq.NewQueueRepository(mqConfig) var msgRepo domain.MessageRepository if mqConfig.Protocol == "amqp" { @@ -52,7 +52,7 @@ func main() { } // 3. Initialize UseCases - uc := usecase.NewActiveMQUseCase(jolokiaClient, msgRepo, cfg.Encoding) + uc := usecase.NewActiveMQUseCase(queueRepo, msgRepo, cfg.Encoding) // 4. Determine ReadOnly state (Flag overrides config) isReadOnly := mqConfig.ReadOnly diff --git a/config.yml b/config.yml index ff02fd1..2625344 100644 --- a/config.yml +++ b/config.yml @@ -6,7 +6,8 @@ encoding: utf-8 environments: dev: - protocol: "stomp" # or "amqp" + broker_type: "auto" # "auto", "classic", or "artemis" (default: auto) + protocol: "stomp" # "stomp" or "amqp" host: "${MQ_HOST:-127.0.0.1}" stomp_port: "61613" # optional (default: 61613) web_port: "8161" # optional (default: 8161) @@ -14,6 +15,7 @@ environments: password: "${MQ_PASS:-admin}" readonly: false prod: + broker_type: "auto" protocol: "amqp" host: "10.0.0.5" amqp_port: "5672" # optional (default: 5672) diff --git a/config/config.go b/config/config.go index e2b41b1..9134e8c 100644 --- a/config/config.go +++ b/config/config.go @@ -18,17 +18,19 @@ type Config struct { } type ActiveMQConfig struct { - Protocol string `yaml:"protocol"` // "stomp" or "amqp" - Host string `yaml:"host"` // e.g. "1.234.25.133" - ReadOnly bool `yaml:"readonly"` // environment specific readonly override - StompPort string `yaml:"stomp_port"` // default "61613" - AmqpPort string `yaml:"amqp_port"` // default "5672" - WebPort string `yaml:"web_port"` // default "8161" - StompURL string `yaml:"stomp_url"` // optional full override - AmqpURL string `yaml:"amqp_url"` // optional full override - JolokiaURL string `yaml:"jolokia_url"` // optional full override - Username string `yaml:"username"` - Password string `yaml:"password"` // #nosec G117 -- plain config field, not a leaked secret + BrokerType string `yaml:"broker_type"` // "auto", "classic", or "artemis" (default "auto") + Protocol string `yaml:"protocol"` // "stomp" or "amqp" + Host string `yaml:"host"` // e.g. "1.234.25.133" + ReadOnly bool `yaml:"readonly"` // environment specific readonly override + StompPort string `yaml:"stomp_port"` // default "61613" + AmqpPort string `yaml:"amqp_port"` // default "5672" + WebPort string `yaml:"web_port"` // default "8161" + StompURL string `yaml:"stomp_url"` // optional full override + AmqpURL string `yaml:"amqp_url"` // optional full override + JolokiaURL string `yaml:"jolokia_url"` // optional full override + JolokiaPath string `yaml:"jolokia_path"` // optional path override (e.g. "/console/jolokia") + Username string `yaml:"username"` + Password string `yaml:"password"` // #nosec G117 -- plain config field, not a leaked secret } func LoadConfig(path string) (*Config, error) { @@ -102,6 +104,9 @@ func parseConfig(data []byte) (*Config, error) { for key, mq := range cfg.Environments { if mq.Host != "" { + if mq.BrokerType == "" { + mq.BrokerType = "auto" + } if mq.Protocol == "" { mq.Protocol = "stomp" } @@ -124,7 +129,13 @@ func parseConfig(data []byte) (*Config, error) { mq.AmqpURL = fmt.Sprintf("amqp://%s:%s", mq.Host, amqpPort) } if mq.JolokiaURL == "" { - mq.JolokiaURL = fmt.Sprintf("http://%s:%s/api/jolokia", mq.Host, webPort) + if mq.JolokiaPath != "" { + mq.JolokiaURL = fmt.Sprintf("http://%s:%s%s", mq.Host, webPort, mq.JolokiaPath) + } else if strings.EqualFold(mq.BrokerType, "artemis") { + mq.JolokiaURL = fmt.Sprintf("http://%s:%s/console/jolokia", mq.Host, webPort) + } else if strings.EqualFold(mq.BrokerType, "classic") { + mq.JolokiaURL = fmt.Sprintf("http://%s:%s/api/jolokia", mq.Host, webPort) + } } cfg.Environments[key] = mq } @@ -155,6 +166,7 @@ encoding: "utf-8" environments: dev: + broker_type: auto protocol: stomp host: 127.0.0.1 stomp_port: 61613 @@ -164,6 +176,7 @@ environments: password: admin readonly: false prod: + broker_type: auto protocol: amqp host: 192.168.0.100 stomp_port: 61613 diff --git a/scratch/inspect_conns.go b/scratch/inspect_conns.go new file mode 100644 index 0000000..0b2e27d --- /dev/null +++ b/scratch/inspect_conns.go @@ -0,0 +1,18 @@ +package main + +import ( + "fmt" + "github.com/xvlet/amqcli/adapter/outbound/activemq" + "github.com/xvlet/amqcli/config" +) + +func main() { + cfg, _ := config.LoadConfig("config.yml") + mqConfig := cfg.Environments["dev"] + client := activemq.NewArtemisJolokiaClient(mqConfig) + _ = client + + // Check ServerControl operations: listSessionsAsJSON + // Also check connection MBeans + fmt.Println("Inspecting Artemis connection and session details...") +} diff --git a/usecase/usecase_test.go b/usecase/usecase_test.go new file mode 100644 index 0000000..7359fa8 --- /dev/null +++ b/usecase/usecase_test.go @@ -0,0 +1,102 @@ +package usecase + +import ( + "testing" + "time" + + "github.com/xvlet/amqcli/domain" +) + +type mockQueueRepo struct { + queues []domain.Queue + detail *domain.QueueDetail +} + +func (m *mockQueueRepo) GetBrokerStats() (domain.BrokerStats, error) { + return domain.BrokerStats{TotalEnqueueCount: 100}, nil +} +func (m *mockQueueRepo) GetBrokerInfo() (string, error) { + return "Apache ActiveMQ Artemis 2.33.0", nil +} +func (m *mockQueueRepo) GetJVMStats() (domain.JVMStats, error) { + return domain.JVMStats{ThreadCount: 10}, nil +} +func (m *mockQueueRepo) GetQueues() ([]domain.Queue, error) { + return m.queues, nil +} +func (m *mockQueueRepo) GetTopics() ([]domain.Topic, error) { + return []domain.Topic{{Name: "TOPIC.1"}}, nil +} +func (m *mockQueueRepo) GetQueueDetail(name string) (*domain.QueueDetail, error) { + return m.detail, nil +} +func (m *mockQueueRepo) GetConnections() ([]domain.Connection, error) { + return []domain.Connection{{Name: "conn-1"}}, nil +} +func (m *mockQueueRepo) GetAllConsumers() ([]domain.Consumer, error) { + return []domain.Consumer{{ConsumerID: "c1"}}, nil +} +func (m *mockQueueRepo) CreateQueue(name string) error { return nil } +func (m *mockQueueRepo) DeleteQueue(name string) error { return nil } +func (m *mockQueueRepo) PurgeQueue(name string) error { return nil } +func (m *mockQueueRepo) RemoveMessage(queueName string, messageID string) error { + return nil +} +func (m *mockQueueRepo) MoveMessage(queueName string, messageID string, destQueue string) error { + return nil +} +func (m *mockQueueRepo) CopyMessage(queueName string, messageID string, destQueue string) error { + return nil +} +func (m *mockQueueRepo) RetryMessage(queueName string, messageID string) error { + return nil +} + +type mockMessageRepo struct{} + +func (m *mockMessageRepo) BrowseQueue(queueName string) ([]domain.Message, error) { + return []domain.Message{{MessageID: "1", Body: "hello"}}, nil +} +func (m *mockMessageRepo) BrowseQueueWithPagination(queueName string, limit int, selector string) ([]domain.Message, error) { + return []domain.Message{{MessageID: "1", Body: "hello"}}, nil +} +func (m *mockMessageRepo) SendMessage(queueName string, correlationID string, ttl time.Duration, body string) error { + return nil +} +func (m *mockMessageRepo) BrowseMessagesByCorrelationID(queueName string, correlationID string) ([]domain.Message, error) { + return []domain.Message{{MessageID: "1", CorrelationID: correlationID}}, nil +} +func (m *mockMessageRepo) DeleteMessagesByCorrelationID(queueName string, correlationID string) error { + return nil +} +func (m *mockMessageRepo) DeleteMessagesBySelector(queueName string, selector string) error { + return nil +} +func (m *mockMessageRepo) ConsumeMessageDestructive(queueName string) (string, error) { + return "test message content", nil +} + +func TestActiveMQUseCase_ArtemisAndClassic(t *testing.T) { + qRepo := &mockQueueRepo{ + queues: []domain.Queue{{Name: "QUEUE.1", Pending: 10}}, + detail: &domain.QueueDetail{Name: "QUEUE.1", QueueSize: 10}, + } + mRepo := &mockMessageRepo{} + + uc := NewActiveMQUseCase(qRepo, mRepo, "utf-8") + + info, err := uc.GetBrokerInfo() + if err != nil || info != "Apache ActiveMQ Artemis 2.33.0" { + t.Fatalf("GetBrokerInfo failed: %v, got %s", err, info) + } + + queues, err := uc.GetQueues() + if err != nil || len(queues) != 1 { + t.Fatalf("GetQueues failed: %v", err) + } + + fullBody, err := uc.GetFullMessageBody("QUEUE.1", "1") + if err != nil || fullBody != "test message content" { + t.Fatalf("GetFullMessageBody failed: %v, got %s", err, fullBody) + } +}