Microservices Communication Patterns: Complete Guide

// REST API Client with proper error handling
class UserServiceClient {
    constructor(baseUrl, options = {}) {
        this.baseUrl = baseUrl;
        this.timeout = options.timeout || 5000;
        this.retryCount = options.retryCount || 3;
    }
    
    async getUser(userId) {
        const response = await fetch(`${this.baseUrl}/users/${userId}`, {
            method: 'GET',
            headers: {
                'Content-Type': 'application/json',
                'Authorization': `Bearer ${this.token}`
            },
            signal: AbortSignal.timeout(this.timeout)
        });
        
        if (!response.ok) {
            throw new ApiError(response.status, await response.text());
        }
        
        return response.json();
    }
    
    async createUser(userData) {
        const response = await fetch(`${this.baseUrl}/users`, {
            method: 'POST',
            headers: {
                'Content-Type': 'application/json'
            },
            body: JSON.stringify(userData)
        });
        
        return response.json();
    }
    
    async withRetry(fn) {
        let lastError;
        for (let i = 0; i < this.retryCount; i++) {
            try {
                return await fn();
            } catch (error) {
                lastError = error;
                if (!this.isRetryable(error)) throw error;
                await this.delay(Math.pow(2, i));
            }
        }
        throw lastError;
    }
    
    isRetryable(error) {
        return error.status >= 500 || error.status === 429;
    }
    
    delay(ms) {
        return new Promise(resolve => setTimeout(resolve, ms));
    }
}

class ApiError extends Error {
    constructor(status, message) {
        super(message);
        this.status = status;
    }
}

// user.proto
syntax = "proto3";

package user;

service UserService {
    rpc GetUser (GetUserRequest) returns (User);
    rpc CreateUser (CreateUserRequest) returns (User);
    rpc StreamUsers (StreamRequest) returns (stream User);
    rpc UpdateUser (User) returns (User);
}

message GetUserRequest {
    string user_id = 1;
}

message CreateUserRequest {
    string name = 1;
    string email = 2;
}

message User {
    string id = 1;
    string name = 2;
    string email = 3;
    int64 created_at = 4;
}

// Go gRPC server implementation
package main

import (
    "context"
    "net"
    
    "google.golang.org/grpc"
    pb "your-package/user"
)

type UserServer struct {
    pb.UnimplementedUserServiceServer
    store UserStore
}

func (s *UserServer) GetUser(ctx context.Context, req *pb.GetUserRequest) (*pb.User, error) {
    user, err := s.store.Get(ctx, req.UserId)
    if err != nil {
        return nil, err
    }
    return &pb.User{
        Id:        user.ID,
        Name:      user.Name,
        Email:     user.Email,
        CreatedAt: user.CreatedAt,
    }, nil
}

func (s *UserServer) StreamUsers(req *pb.StreamRequest, stream pb.UserService_StreamUsersServer) error {
    users, err := s.store.List(stream.Context())
    if err != nil {
        return err
    }
    
    for _, user := range users {
        if err := stream.Send(&pb.User{
            Id:    user.ID,
            Name:  user.Name,
            Email: user.Email,
        }); err != nil {
            return err
        }
    }
    
    return nil
}

// RabbitMQ Producer
const amqp = require('amqplib');

class MessageProducer {
    constructor(url, queueName) {
        this.url = url;
        this.queueName = queueName;
        this.connection = null;
        this.channel = null;
    }
    
    async connect() {
        this.connection = await amqp.connect(this.url);
        this.channel = await this.connection.createChannel();
        await this.channel.assertQueue(this.queueName, { durable: true });
    }
    
    async publish(message, options = {}) {
        const messageBuffer = Buffer.from(JSON.stringify(message));
        
        return this.channel.sendToQueue(
            this.queueName,
            messageBuffer,
            {
                persistent: true,
                correlationId: options.correlationId,
                replyTo: options.replyTo,
                ...options.headers
            }
        );
    }
    
    async publishWithReply(message, replyQueue) {
        const correlationId = crypto.randomUUID();
        
        return new Promise((resolve, reject) => {
            const timeout = setTimeout(() => {
                reject(new Error('Request timeout'));
            }, 5000);
            
            this.channel.consume(replyQueue, (msg) => {
                if (msg.properties.correlationId === correlationId) {
                    clearTimeout(timeout);
                    resolve(JSON.parse(msg.content.toString()));
                }
            }, { noAck: true });
            
            this.publish(message, { correlationId, replyTo: replyQueue });
        });
    }
}

// RabbitMQ Consumer
class MessageConsumer {
    constructor(url, queueName) {
        this.url = url;
        this.queueName = queueName;
    }
    
    async start(handler) {
        const connection = await amqp.connect(this.url);
        const channel = await connection.createChannel();
        await channel.assertQueue(this.queueName, { durable: true });
        
        channel.prefetch(10);
        
        channel.consume(this.queueName, async (msg) => {
            if (!msg) return;
            
            try {
                const content = JSON.parse(msg.content.toString());
                await handler(content);
                channel.ack(msg);
            } catch (error) {
                console.error('Processing failed:', error);
                channel.nack(msg, false, true);
            }
        });
    }
}

// Event Bus Implementation
interface Event {
    type: string;
    payload: any;
    metadata: {
        timestamp: number;
        correlationId: string;
        causationId?: string;
    };
}

class EventBus {
    private handlers: Map<string, EventHandler[]> = new Map();
    private eventStore: EventStore;
    
    async publish(event: Event): Promise<void> {
        // Store event for replay
        await this.eventStore.save(event);
        
        // Get handlers for this event type
        const handlers = this.handlers.get(event.type) || [];
        
        // Execute handlers (can be async)
        await Promise.allSettled(
            handlers.map(handler => handler(event))
        );
    }
    
    subscribe(eventType: string, handler: EventHandler): void {
        if (!this.handlers.has(eventType)) {
            this.handlers.set(eventType, []);
        }
        this.handlers.get(eventType).push(handler);
    }
}

// Example: UserCreated event
class UserCreatedEventHandler implements EventHandler {
    async handle(event: Event): Promise<void> {
        const { userId, email, name } = event.payload;
        
        // Send welcome email
        await this.emailService.send({
            to: email,
            template: 'welcome',
            data: { name }
        });
        
        // Create default workspace
        await this.workspaceService.createDefault(userId);
        
        // Notify analytics
        await this.analytics.track('user_created', { userId });
    }
}

# API Gateway Configuration (Kong)
services:
  - name: user-service
    url: http://user-service:8080
    routes:
      - name: user-routes
        paths:
          - /api/users
        methods:
          - GET
          - POST
    plugins:
      - name: rate-limiting
        config:
          minute: 100
          policy: local
      - name: jwt
      - name: correlation-id
      
  - name: order-service
    url: http://order-service:8080
    routes:
      - name: order-routes
        paths:
          - /api/orders
    plugins:
      - name: rate-limiting
        config:
          minute: 50
      - name: jwt

# Upstream configuration for load balancing
upstreams:
  - name: user-service-upstream
    targets:
      - target: user-service-1:8080
        weight: 100
      - target: user-service-2:8080
        weight: 100
    healthchecks:
      active:
        type: http
        http_path: /health
        interval: 5
        unhealthy: 3

apiVersion: networking.istio.io/v1beta1
kind: VirtualService
metadata:
  name: reviews
spec:
  hosts:
    - reviews
  http:
    - match:
        - headers:
            end-user:
              exact: jason
      route:
        - destination:
            host: reviews
            subset: v2
    - route:
        - destination:
            host: reviews
            subset: v1

---
apiVersion: networking.istio.io/v1beta1
kind: DestinationRule
metadata:
  name: reviews
spec:
  host: reviews
  trafficPolicy:
    connectionPool:
      tcp:
        maxConnections: 100
      http:
        h2UpgradePolicy: UPGRADE
        http1MaxPendingRequests: 100
        http2MaxRequests: 1000
    loadBalancer:
      simple: LEAST_REQUEST
    outlierDetection:
      consecutive5xxErrors: 5
      interval: 30s
      baseEjectionTime: 30s