Publish-Subscribe Pattern

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

// Designing filterable events with attributes
 
interface FilterableMessage<T> {
  // Payload: The actual event data (not used for filtering)
  data: T;
  
  // Attributes: Metadata for filtering (indexed, inspectable by broker)
  attributes: {
    // Event classification
    eventType: string;           // 'ORDER_CREATED', 'ORDER_SHIPPED', etc.
    eventVersion: string;        // 'v1.2' - for schema versioning
    
    // Domain categorization
    domain: string;              // 'orders', 'inventory', 'payments'
    aggregate: string;           // 'order', 'product', 'customer'
    aggregateId: string;         // Specific entity ID
    
    // Common filter dimensions
    region: string;              // 'us-east', 'eu-west', 'apac'
    environment: string;         // 'production', 'staging', 'development'
    priority: string;            // 'high', 'normal', 'low'
    
    // Business-specific dimensions
    amountTier?: string;         // 'high-value', 'standard', 'micro'
    customerTier?: string;       // 'enterprise', 'premium', 'free'
    channel?: string;            // 'web', 'mobile', 'api'
  };
  
  // System metadata (not for filtering, but useful)
  messageId: string;
  timestamp: string;
  correlationId?: string;
}
 
// Example: Publishing with filter-friendly attributes
async function publishOrderCreated(order: Order): Promise<void> {
  const message: FilterableMessage<OrderCreatedPayload> = {
    data: {
      orderId: order.id,
      customerId: order.customerId,
      items: order.items,
      totalAmount: order.totalAmount,
      // ... full payload
    },
    attributes: {
      eventType: 'ORDER_CREATED',
      eventVersion: 'v2.1',
      domain: 'orders',
      aggregate: 'order',
      aggregateId: order.id,
      region: order.shippingAddress.region,
      environment: process.env.ENVIRONMENT || 'production',
      priority: order.isExpress ? 'high' : 'normal',
      amountTier: order.totalAmount > 1000 ? 'high-value' : 
                  order.totalAmount > 100 ? 'standard' : 'micro',
      customerTier: order.customer.tier,
      channel: order.sourceChannel,
    },
    messageId: generateId(),
    timestamp: new Date().toISOString(),
    correlationId: order.sessionId,
  };
  
  await broker.publish('orders', message);
}

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

// Content-Based Router Implementation
 
interface RoutingRule {
  name: string;
  condition: (event: any) => boolean;
  destinations: string[];
  priority: number; // Lower = higher priority
  exclusive: boolean; // If true, stop after this rule matches
}
 
class ContentBasedRouter {
  private rules: RoutingRule[] = [];
 
  constructor(rulesConfig: RoutingRule[]) {
    // Sort by priority
    this.rules = rulesConfig.sort((a, b) => a.priority - b.priority);
  }
 
  async route(message: Message): Promise<void> {
    const event = JSON.parse(message.data);
    const destinations = new Set<string>();
 
    for (const rule of this.rules) {
      try {
        if (rule.condition(event)) {
          rule.destinations.forEach(d => destinations.add(d));
          
          if (rule.exclusive) {
            break; // Stop evaluating rules
          }
        }
      } catch (error) {
        // Rule evaluation error - log and continue
        this.metrics.ruleEvaluationError(rule.name, error);
      }
    }
 
    // Default destination if no rules matched
    if (destinations.size === 0) {
      destinations.add('default-handler');
    }
 
    // Publish to all matched destinations
    await Promise.all(
      Array.from(destinations).map(dest =>
        this.broker.publish(dest, message.data, message.attributes)
      )
    );
 
    await message.ack();
  }
}
 
// Example routing rules
const orderRoutingRules: RoutingRule[] = [
  {
    name: 'hazmat-detection',
    condition: (event) => event.items.some((item: any) => 
      item.categories?.includes('hazardous')
    ),
    destinations: ['orders.hazmat'],
    priority: 10,
    exclusive: false, // Hazmat orders may also match other rules
  },
  {
    name: 'vip-customer',
    condition: (event) => event.customer.tier === 'enterprise' || 
                          event.customer.lifetimeValue > 100000,
    destinations: ['orders.vip', 'orders.analytics.vip'],
    priority: 20,
    exclusive: false,
  },
  {
    name: 'bulk-order',
    condition: (event) => event.items.length > 10 || 
                          event.totalQuantity > 100,
    destinations: ['orders.bulk'],
    priority: 30,
    exclusive: true, // Bulk orders get special handling, skip other rules
  },
  {
    name: 'express-shipping',
    condition: (event) => event.shipping.method === 'express' ||
                          event.shipping.method === 'same-day',
    destinations: ['orders.priority'],
    priority: 40,
    exclusive: false,
  },
];

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

// Filter Optimization Strategies
 
// ❌ BAD: Complex filter that's hard to evaluate
const inefficientFilter = `
  (attributes.eventType = "A" OR attributes.eventType = "B" OR 
   attributes.eventType = "C" OR attributes.eventType = "D") AND
  (attributes.region = "us-east" OR attributes.region = "us-west" OR
   attributes.region = "eu-west") AND
  hasPrefix(attributes.category, "electronics") AND
  NOT attributes.environment = "development"
`;
 
// ✅ BETTER: Simplified with category attribute
// Publisher creates a derived attribute for common filter patterns
const efficientFilter = `
  attributes.filterCategory = "electronics-prod-us-eu" AND
  attributes.eventType IN ("A", "B", "C", "D")
`;
 
// ✅ BEST: Split into topic hierarchy if volume warrants
// Topic: orders.electronics.prod.{region}
// Subscribe to: orders.electronics.prod.us-east, orders.electronics.prod.us-west, ...
 
// For Kafka: Pre-filter with Kafka Streams into derived topics
// Input topic: all-orders (high volume)
// Kafka Streams job filters and routes to:
//   - orders.electronics.us
//   - orders.electronics.eu
//   - orders.clothing.us
// Consumers subscribe to specific filtered topics

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

// Unit testing filter expressions
 
import { parseGooglePubSubFilter, evaluateFilter } from './filter-parser';
 
describe('Order filter expressions', () => {
  const filter = parseGooglePubSubFilter(
    'attributes.eventType = "ORDER_CREATED" AND attributes.region = "eu-west"'
  );
 
  it('matches EU order created events', () => {
    const event = {
      attributes: { eventType: 'ORDER_CREATED', region: 'eu-west' }
    };
    expect(evaluateFilter(filter, event)).toBe(true);
  });
 
  it('rejects non-EU regions', () => {
    const event = {
      attributes: { eventType: 'ORDER_CREATED', region: 'us-east' }
    };
    expect(evaluateFilter(filter, event)).toBe(false);
  });
 
  it('rejects wrong event types', () => {
    const event = {
      attributes: { eventType: 'ORDER_SHIPPED', region: 'eu-west' }
    };
    expect(evaluateFilter(filter, event)).toBe(false);
  });
 
  it('rejects missing attributes', () => {
    const event = {
      attributes: { eventType: 'ORDER_CREATED' } // missing region
    };
    expect(evaluateFilter(filter, event)).toBe(false);
  });
});
 
// Integration test: Verify filter delivery
describe('Filter delivery integration', () => {
  it('delivers only matching events to filtered subscription', async () => {
    // Create subscription with filter
    const sub = await createSubscription('test-orders', {
      filter: 'attributes.priority = "high"'
    });
    
    // Publish mix of events
    await publish('test-orders', { data: 'high-1', attributes: { priority: 'high' } });
    await publish('test-orders', { data: 'low-1', attributes: { priority: 'low' } });
    await publish('test-orders', { data: 'high-2', attributes: { priority: 'high' } });
    
    // Wait and collect
    const received = await collectMessages(sub, { timeout: 5000 });
    
    // Verify only high priority received
    expect(received).toHaveLength(2);
    expect(received.every(m => m.attributes.priority === 'high')).toBe(true);
  });
});