activemq

What this skill does

# Apache ActiveMQ Core Knowledge

> **Deep Knowledge**: Use `mcp__documentation__fetch_docs` with technology: `activemq` for comprehensive documentation.

## Quick Start (Docker)

```yaml
# docker-compose.yml
services:
  activemq:
    image: apache/activemq-artemis:latest
    ports:
      - "61616:61616" # AMQP/OpenWire
      - "8161:8161"   # Web Console
    environment:
      - ARTEMIS_USER=admin
      - ARTEMIS_PASSWORD=admin
    volumes:
      - activemq_data:/var/lib/artemis-instance

volumes:
  activemq_data:
```

```bash
docker-compose up -d
# Web Console: http://localhost:8161/console
```

## Core Concepts

| Concept | Description |
|---------|-------------|
| **Queue** | Point-to-point messaging |
| **Topic** | Publish-subscribe messaging |
| **Selector** | SQL-like message filtering |
| **Durable Subscriber** | Persisted topic subscription |
| **Message Groups** | Ordered message processing |
| **Virtual Topic** | Queue semantics on topics |

## JMS Architecture

```
┌─────────────────────────────────────────────────────────┐
│                    ActiveMQ Broker                       │
│                                                         │
│   Queue (P2P)              Topic (Pub/Sub)             │
│   ┌─────────┐              ┌─────────┐                 │
│   │ Message │              │ Message │                 │
│   │  Queue  │──▶Consumer   │  Topic  │──▶Subscriber 1 │
│   └─────────┘              └─────────┘──▶Subscriber 2 │
│        ▲                        ▲                      │
│        │                        │                      │
│   Producer                  Publisher                  │
└─────────────────────────────────────────────────────────┘
```

## JMS Message Types

| Type | Description |
|------|-------------|
| `TextMessage` | String content |
| `BytesMessage` | Binary data |
| `MapMessage` | Key-value pairs |
| `ObjectMessage` | Serialized Java object |
| `StreamMessage` | Sequential data stream |

## Producer Patterns

### Java (Spring JMS)
```java
@Configuration
@EnableJms
public class JmsConfig {
    @Bean
    public ConnectionFactory connectionFactory() {
        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory();
        factory.setBrokerURL("tcp://localhost:61616");
        factory.setUserName("admin");
        factory.setPassword("admin");
        return factory;
    }

    @Bean
    public JmsTemplate jmsTemplate(ConnectionFactory connectionFactory) {
        JmsTemplate template = new JmsTemplate(connectionFactory);
        template.setDeliveryPersistent(true);
        template.setSessionAcknowledgeMode(Session.CLIENT_ACKNOWLEDGE);
        return template;
    }

    @Bean
    public JmsTemplate topicJmsTemplate(ConnectionFactory connectionFactory) {
        JmsTemplate template = new JmsTemplate(connectionFactory);
        template.setPubSubDomain(true);  // Enable topic mode
        return template;
    }
}

@Service
public class OrderProducer {
    @Autowired
    private JmsTemplate jmsTemplate;

    // Send to queue
    public void sendToQueue(Order order) {
        jmsTemplate.convertAndSend("orders.queue", order, message -> {
            message.setJMSCorrelationID(UUID.randomUUID().toString());
            message.setStringProperty("orderType", order.getType());
            message.setIntProperty("priority", order.getPriority());
            return message;
        });
    }

    // Send with reply
    public OrderResponse sendAndReceive(Order order) {
        return (OrderResponse) jmsTemplate.sendAndReceive("orders.queue",
            session -> {
                ObjectMessage msg = session.createObjectMessage(order);
                msg.setJMSReplyTo(session.createTemporaryQueue());
                return msg;
            });
    }
}

@Service
public class EventPublisher {
    @Autowired
    @Qualifier("topicJmsTemplate")
    private JmsTemplate topicTemplate;

    // Publish to topic
    public void publishEvent(OrderEvent event) {
        topicTemplate.convertAndSend("orders.events", event);
    }
}
```

### Node.js (stompit)
```typescript
import stompit from 'stompit';

const connectOptions = {
  host: 'localhost',
  port: 61613,
  connectHeaders: {
    host: '/',
    login: 'admin',
    passcode: 'admin',
    'heart-beat': '5000,5000',
  },
};

stompit.connect(connectOptions, (error, client) => {
  if (error) {
    console.error('Connection error:', error);
    return;
  }

  const sendHeaders = {
    destination: '/queue/orders',
    'content-type': 'application/json',
    persistent: 'true',
    'correlation-id': uuidv4(),
  };

  const frame = client.send(sendHeaders);
  frame.write(JSON.stringify(order));
  frame.end();

  client.disconnect();
});
```

### Python (stomp.py)
```python
import stomp
import json

class OrderListener(stomp.ConnectionListener):
    def on_error(self, frame):
        print(f'Error: {frame.body}')

    def on_message(self, frame):
        print(f'Received: {frame.body}')

conn = stomp.Connection([('localhost', 61613)])
conn.set_listener('', OrderListener())
conn.connect('admin', 'admin', wait=True)

# Send to queue
conn.send(
    destination='/queue/orders',
    body=json.dumps(order),
    headers={
        'persistent': 'true',
        'content-type': 'application/json',
        'correlation-id': correlation_id,
    }
)

conn.disconnect()
```

## Consumer Patterns

### Java (Spring JMS)
```java
@Configuration
public class JmsListenerConfig {
    @Bean
    public DefaultJmsListenerContainerFactory jmsListenerContainerFactory(
            ConnectionFactory connectionFactory) {
        DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setConcurrency("3-10");
        factory.setSessionAcknowledgeMode(Session.CLIENT_ACKNOWLEDGE);
        factory.setErrorHandler(t -> log.error("JMS Error", t));
        return factory;
    }

    @Bean
    public DefaultJmsListenerContainerFactory topicListenerContainerFactory(
            ConnectionFactory connectionFactory) {
        DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setPubSubDomain(true);
        factory.setSubscriptionDurable(true);
        factory.setClientId("order-service");
        return factory;
    }
}

@Service
public class OrderConsumer {
    // Queue consumer
    @JmsListener(destination = "orders.queue", concurrency = "3-10")
    public void consumeQueue(
            @Payload Order order,
            @Header(JmsHeaders.CORRELATION_ID) String correlationId,
            @Header(name = "orderType", required = false) String orderType,
            Session session,
            Message message) throws JMSException {

        try {
            processOrder(order);
            message.acknowledge();
        } catch (Exception e) {
            session.recover();  // Redelivery
        }
    }

    // Queue consumer with selector
    @JmsListener(
        destination = "orders.queue",
        selector = "orderType = 'EXPRESS' AND priority > 5"
    )
    public void consumeExpressOrders(Order order) {
        processExpressOrder(order);
    }

    // Durable topic subscriber
    @JmsListener(
        destination = "orders.events",
        containerFactory = "topicListenerContainerFactory",
        subscription = "order-processor"
    )
    public void consumeTopic(OrderEvent event) {
        processEvent(event);
    }
}
```

### Request/Reply Pattern
```java
@Service
public class OrderService {
    @JmsListener(destination = "orders.request")
    @SendTo("orders.response")
    public OrderResponse processOrderRequest(Order order) {
        // Process and return response
        return new OrderResponse(order.getId(), "PROCESSED");
    }
}
```

### Node.js (stompit)
```typescript
stompit.connect(connectOptions, (error, client) => {
  const subscribeHeaders = {
    destination: '/qu