Skip to main content

Server Development

Creating Hypermodern Servers

Building a Hypermodern server means creating an application that can simultaneously handle HTTP, WebSocket, and TCP connections while sharing the same business logic across all protocols.

Basic Server Setup

import 'package:hypermodern_server/hypermodern_server.dart';
import 'generated/models.dart';
import 'generated/server.dart';

void main() async {
  final server = HypermodernServer();
  
  // Configure server
  server.configure(ServerConfig(
    // Protocol ports
    httpPort: 8080,
    wsPort: 8082,
    tcpPort: 8081,
    
    // Performance settings
    maxConnections: 1000,
    requestTimeout: Duration(seconds: 30),
    keepAliveTimeout: Duration(seconds: 60),
    
    // Security
    corsEnabled: true,
    corsOrigins: ['http://localhost:3000', 'https://myapp.com'],
    rateLimitEnabled: true,
    rateLimitRequests: 100,
    rateLimitWindow: Duration(minutes: 1),
  ));
  
  // Register endpoints
  _registerEndpoints(server);
  
  // Start all protocol servers
  await server.listen();
  
  print('🚀 Hypermodern server running on:');
  print('   HTTP:      http://localhost:8080');
  print('   WebSocket: ws://localhost:8082');
  print('   TCP:       localhost:8081');
}

Server Configuration Options

class ServerConfig {
  // Protocol configuration
  final int httpPort;
  final int wsPort;
  final int tcpPort;
  final String bindAddress;
  
  // Performance tuning
  final int maxConnections;
  final int maxRequestSize;
  final Duration requestTimeout;
  final Duration keepAliveTimeout;
  final int workerThreads;
  
  // Security settings
  final bool corsEnabled;
  final List<String> corsOrigins;
  final bool rateLimitEnabled;
  final int rateLimitRequests;
  final Duration rateLimitWindow;
  final bool validateRequests;
  
  // Logging and monitoring
  final LogLevel logLevel;
  final bool metricsEnabled;
  final String? metricsEndpoint;
  
  // SSL/TLS (for production)
  final String? sslCertPath;
  final String? sslKeyPath;
  final bool requireSsl;
  
  const ServerConfig({
    this.httpPort = 8080,
    this.wsPort = 8082,
    this.tcpPort = 8081,
    this.bindAddress = '0.0.0.0',
    this.maxConnections = 1000,
    this.maxRequestSize = 1024 * 1024, // 1MB
    this.requestTimeout = const Duration(seconds: 30),
    this.keepAliveTimeout = const Duration(seconds: 60),
    this.workerThreads = 4,
    this.corsEnabled = false,
    this.corsOrigins = const [],
    this.rateLimitEnabled = false,
    this.rateLimitRequests = 100,
    this.rateLimitWindow = const Duration(minutes: 1),
    this.validateRequests = true,
    this.logLevel = LogLevel.info,
    this.metricsEnabled = false,
    this.metricsEndpoint,
    this.sslCertPath,
    this.sslKeyPath,
    this.requireSsl = false,
  });
}

Implementing Endpoint Handlers

Endpoint handlers contain your business logic and are automatically available across all protocols.

Basic Endpoint Implementation

void _registerEndpoints(HypermodernServer server) {
  // Simple request-response endpoint
  server.registerEndpoint<GetUserRequest, User>(
    'get_user',
    (request) async {
      final user = await userService.getUserById(request.id);
      if (user == null) {
        throw NotFoundException('User not found', details: {
          'user_id': request.id.toString(),
        });
      }
      return user;
    },
  );
  
  // Endpoint with validation
  server.registerEndpoint<CreateUserRequest, User>(
    'create_user',
    (request) async {
      // Validate request
      if (request.username.length < 3) {
        throw ValidationException('Username too short', fieldErrors: {
          'username': 'Must be at least 3 characters',
        });
      }
      
      if (!_isValidEmail(request.email)) {
        throw ValidationException('Invalid email format', fieldErrors: {
          'email': 'Must be a valid email address',
        });
      }
      
      // Check for existing user
      final existingUser = await userService.getUserByEmail(request.email);
      if (existingUser != null) {
        throw ConflictException('User already exists', details: {
          'email': request.email,
        });
      }
      
      // Create user
      final user = await userService.createUser(
        username: request.username,
        email: request.email,
        passwordHash: _hashPassword(request.password),
      );
      
      return user;
    },
  );
  
  // Endpoint with complex response
  server.registerEndpoint<SearchUsersRequest, SearchUsersResponse>(
    'search_users',
    (request) async {
      final results = await userService.searchUsers(
        query: request.query,
        filters: request.filters,
        pagination: request.pagination,
        sort: request.sort,
      );
      
      return SearchUsersResponse(
        users: results.users,
        totalCount: results.totalCount,
        page: request.pagination.page,
        hasMore: results.hasMore,
        facets: results.facets,
      );
    },
  );
}

Async and Database Operations

class UserService {
  final Database db;
  
  UserService(this.db);
  
  Future<User?> getUserById(int id) async {
    final result = await db.query(
      'SELECT * FROM users WHERE id = ?',
      [id],
    );
    
    if (result.isEmpty) return null;
    
    return User.fromJson(result.first);
  }
  
  Future<User> createUser({
    required String username,
    required String email,
    required String passwordHash,
  }) async {
    final now = DateTime.now();
    
    final result = await db.query(
      '''
      INSERT INTO users (username, email, password_hash, created_at, updated_at)
      VALUES (?, ?, ?, ?, ?)
      RETURNING *
      ''',
      [username, email, passwordHash, now, now],
    );
    
    return User.fromJson(result.first);
  }
  
  Future<SearchResult> searchUsers({
    String? query,
    UserFilters? filters,
    required Pagination pagination,
    required Sort sort,
  }) async {
    final queryBuilder = QueryBuilder('users');
    
    // Add search conditions
    if (query != null && query.isNotEmpty) {
      queryBuilder.where('username ILIKE ? OR email ILIKE ?', 
                        ['%$query%', '%$query%']);
    }
    
    // Add filters
    if (filters?.status != null) {
      queryBuilder.where('status = ?', [filters!.status!.toString()]);
    }
    
    if (filters?.createdAfter != null) {
      queryBuilder.where('created_at > ?', [filters!.createdAfter]);
    }
    
    // Add sorting
    queryBuilder.orderBy('${sort.field} ${sort.direction}');
    
    // Add pagination
    queryBuilder.limit(pagination.limit);
    queryBuilder.offset(pagination.page * pagination.limit);
    
    // Execute queries
    final countQuery = queryBuilder.buildCountQuery();
    final dataQuery = queryBuilder.buildSelectQuery();
    
    final [countResult, dataResult] = await Future.wait([
      db.query(countQuery.sql, countQuery.parameters),
      db.query(dataQuery.sql, dataQuery.parameters),
    ]);
    
    final totalCount = countResult.first['count'] as int;
    final users = dataResult.map((row) => User.fromJson(row)).toList();
    
    return SearchResult(
      users: users,
      totalCount: totalCount,
      hasMore: (pagination.page + 1) * pagination.limit < totalCount,
    );
  }
}

Error Handling in Endpoints

server.registerEndpoint<UpdateUserRequest, User>(
  'update_user',
  (request) async {
    try {
      // Validate permissions
      final currentUser = await getCurrentUser(request.context);
      if (currentUser.id != request.id && !currentUser.isAdmin) {
        throw ForbiddenException('Cannot update other users');
      }
      
      // Get existing user
      final existingUser = await userService.getUserById(request.id);
      if (existingUser == null) {
        throw NotFoundException('User not found');
      }
      
      // Validate updates
      if (request.email != null && !_isValidEmail(request.email!)) {
        throw ValidationException('Invalid email format');
      }
      
      // Check for email conflicts
      if (request.email != null && request.email != existingUser.email) {
        final emailExists = await userService.getUserByEmail(request.email!);
        if (emailExists != null) {
          throw ConflictException('Email already in use');
        }
      }
      
      // Perform update
      final updatedUser = await userService.updateUser(
        id: request.id,
        username: request.username,
        email: request.email,
        profile: request.profile,
      );
      
      // Log the update
      await auditService.logUserUpdate(
        userId: currentUser.id,
        targetUserId: request.id,
        changes: request.getChanges(existingUser),
      );
      
      return updatedUser;
      
    } on DatabaseException catch (e) {
      // Handle database-specific errors
      if (e.isUniqueConstraintViolation) {
        throw ConflictException('Duplicate value detected');
      } else if (e.isForeignKeyViolation) {
        throw ValidationException('Referenced entity does not exist');
      } else {
        throw ServerException('Database operation failed');
      }
    } on ValidationException {
      rethrow; // Re-throw validation errors as-is
    } catch (e) {
      // Log unexpected errors
      logger.error('Unexpected error in update_user', error: e);
      throw ServerException('Internal server error');
    }
  },
);

Middleware and Request Processing

Middleware provides a powerful way to add cross-cutting concerns like authentication, logging, and validation.

Built-in Middleware

void configureMiddleware(HypermodernServer server) {
  // Request logging
  server.middleware.add(LoggingMiddleware(
    logRequests: true,
    logResponses: true,
    logHeaders: false, // Don't log sensitive headers
    logBodies: false,  // Don't log request/response bodies
  ));
  
  // CORS handling
  server.middleware.add(CorsMiddleware(
    allowedOrigins: ['http://localhost:3000', 'https://myapp.com'],
    allowedMethods: ['GET', 'POST', 'PUT', 'DELETE'],
    allowedHeaders: ['Content-Type', 'Authorization'],
    allowCredentials: true,
  ));
  
  // Rate limiting
  server.middleware.add(RateLimitMiddleware(
    requests: 100,
    window: Duration(minutes: 1),
    keyGenerator: (request) => request.clientIp,
  ));
  
  // Request validation
  server.middleware.add(ValidationMiddleware(
    validateRequests: true,
    validateResponses: false, // Disable in production
  ));
  
  // Authentication (applied to specific endpoints)
  server.middleware.add(AuthenticationMiddleware(
    jwtSecret: 'your-secret-key',
    requiredForEndpoints: [
      'create_user',
      'update_user',
      'delete_user',
    ],
  ));
}

Custom Middleware

class TimingMiddleware implements Middleware {
  @override
  Future<dynamic> handle(
    dynamic request,
    Future<dynamic> Function(dynamic) next,
  ) async {
    final stopwatch = Stopwatch()..start();
    
    try {
      final response = await next(request);
      
      final duration = stopwatch.elapsedMilliseconds;
      logger.info('Request completed in ${duration}ms', extra: {
        'endpoint': request.endpoint,
        'duration_ms': duration,
      });
      
      return response;
    } catch (e) {
      final duration = stopwatch.elapsedMilliseconds;
      logger.error('Request failed in ${duration}ms', error: e, extra: {
        'endpoint': request.endpoint,
        'duration_ms': duration,
      });
      rethrow;
    }
  }
}

class CachingMiddleware implements Middleware {
  final Map<String, CacheEntry> _cache = {};
  final Duration defaultTtl;
  
  CachingMiddleware({this.defaultTtl = const Duration(minutes: 5)});
  
  @override
  Future<dynamic> handle(
    dynamic request,
    Future<dynamic> Function(dynamic) next,
  ) async {
    // Only cache GET requests
    if (request.method != 'GET') {
      return await next(request);
    }
    
    final cacheKey = _generateCacheKey(request);
    final entry = _cache[cacheKey];
    
    // Return cached response if valid
    if (entry != null && !entry.isExpired) {
      logger.debug('Cache hit for $cacheKey');
      return entry.data;
    }
    
    // Execute request and cache result
    final response = await next(request);
    
    _cache[cacheKey] = CacheEntry(
      data: response,
      expiresAt: DateTime.now().add(defaultTtl),
    );
    
    logger.debug('Cached response for $cacheKey');
    return response;
  }
  
  String _generateCacheKey(dynamic request) {
    return '${request.endpoint}:${request.hashCode}';
  }
}

class SecurityHeadersMiddleware implements Middleware {
  @override
  Future<dynamic> handle(
    dynamic request,
    Future<dynamic> Function(dynamic) next,
  ) async {
    final response = await next(request);
    
    // Add security headers for HTTP responses
    if (request.protocol == 'http') {
      response.headers.addAll({
        'X-Content-Type-Options': 'nosniff',
        'X-Frame-Options': 'DENY',
        'X-XSS-Protection': '1; mode=block',
        'Strict-Transport-Security': 'max-age=31536000; includeSubDomains',
        'Content-Security-Policy': "default-src 'self'",
      });
    }
    
    return response;
  }
}

Conditional Middleware

Apply middleware only to specific endpoints or conditions:

class ConditionalMiddleware implements Middleware {
  final Middleware middleware;
  final bool Function(dynamic request) condition;
  
  ConditionalMiddleware(this.middleware, this.condition);
  
  @override
  Future<dynamic> handle(
    dynamic request,
    Future<dynamic> Function(dynamic) next,
  ) async {
    if (condition(request)) {
      return await middleware.handle(request, next);
    } else {
      return await next(request);
    }
  }
}

// Usage
server.middleware.add(ConditionalMiddleware(
  AuthenticationMiddleware(jwtSecret: 'secret'),
  (request) => ['create_user', 'update_user'].contains(request.endpoint),
));

server.middleware.add(ConditionalMiddleware(
  CachingMiddleware(defaultTtl: Duration(hours: 1)),
  (request) => request.method == 'GET' && request.endpoint.startsWith('get_'),
));

Streaming Endpoints

Streaming endpoints enable real-time communication and are automatically available over WebSocket and TCP protocols.

Basic Streaming

void registerStreamingEndpoints(HypermodernServer server) {
  // Server-to-client streaming
  server.registerStreamingEndpoint<WatchUsersRequest, UserUpdate>(
    'watch_users',
    (request) async* {
      // Validate request
      if (request.userIds.isEmpty) {
        throw ValidationException('At least one user ID required');
      }
      
      // Create stream controller
      final controller = StreamController<UserUpdate>();
      
      // Set up database change listener
      final subscription = userService.watchUsers(request.userIds).listen(
        (update) => controller.add(update),
        onError: (error) => controller.addError(error),
        onDone: () => controller.close(),
      );
      
      // Clean up when client disconnects
      controller.onCancel = () {
        subscription.cancel();
      };
      
      yield* controller.stream;
    },
  );
  
  // Bidirectional streaming (chat example)
  server.registerBidirectionalStreamingEndpoint<ChatMessage, ChatMessage>(
    'chat_messages',
    (incomingMessages, outgoingMessages) async {
      final roomId = await _extractRoomId(incomingMessages);
      
      // Join chat room
      await chatService.joinRoom(roomId);
      
      // Forward incoming messages to chat service
      final incomingSubscription = incomingMessages.listen(
        (message) async {
          // Validate and process message
          final processedMessage = await chatService.processMessage(message);
          
          // Broadcast to all room members
          await chatService.broadcastMessage(roomId, processedMessage);
        },
        onError: (error) => logger.error('Chat message error', error: error),
      );
      
      // Forward room messages to client
      final outgoingSubscription = chatService.getRoomMessages(roomId).listen(
        (message) => outgoingMessages.add(message),
        onError: (error) => outgoingMessages.addError(error),
      );
      
      // Clean up when connection closes
      outgoingMessages.onCancel = () async {
        await incomingSubscription.cancel();
        await outgoingSubscription.cancel();
        await chatService.leaveRoom(roomId);
      };
    },
  );
}

Advanced Streaming Patterns

// Stream with filtering and transformation
server.registerStreamingEndpoint<WatchOrdersRequest, OrderUpdate>(
  'watch_orders',
  (request) async* {
    final userId = await getCurrentUserId(request.context);
    
    // Create filtered stream
    final orderStream = orderService
        .watchAllOrders()
        .where((order) => order.userId == userId)
        .where((order) => request.statuses.isEmpty || 
                         request.statuses.contains(order.status))
        .map((order) => OrderUpdate.fromOrder(order));
    
    // Add rate limiting to prevent overwhelming clients
    final rateLimitedStream = orderStream.throttle(Duration(milliseconds: 100));
    
    yield* rateLimitedStream;
  },
);

// Stream with backpressure handling
server.registerStreamingEndpoint<WatchMetricsRequest, MetricUpdate>(
  'watch_metrics',
  (request) async* {
    final controller = StreamController<MetricUpdate>();
    
    // High-frequency metrics source
    final metricsSubscription = metricsService.getMetricsStream().listen(
      (metric) {
        // Handle backpressure
        if (controller.hasListener && !controller.isPaused) {
          controller.add(MetricUpdate.fromMetric(metric));
        }
      },
      onError: (error) => controller.addError(error),
    );
    
    // Pause/resume based on client backpressure
    controller.onPause = () => metricsSubscription.pause();
    controller.onResume = () => metricsSubscription.resume();
    controller.onCancel = () => metricsSubscription.cancel();
    
    yield* controller.stream;
  },
);

// Stream with periodic updates
server.registerStreamingEndpoint<WatchSystemStatusRequest, SystemStatus>(
  'watch_system_status',
  (request) async* {
    while (true) {
      try {
        final status = await systemService.getCurrentStatus();
        yield status;
        
        // Wait for next update interval
        await Future.delayed(Duration(seconds: request.intervalSeconds));
      } catch (e) {
        // Log error but continue streaming
        logger.error('Error getting system status', error: e);
        
        // Send error status
        yield SystemStatus(
          healthy: false,
          error: e.toString(),
          timestamp: DateTime.now(),
        );
        
        // Wait before retrying
        await Future.delayed(Duration(seconds: 5));
      }
    }
  },
);

Stream Connection Management

class StreamManager {
  final Map<String, Set<StreamSubscription>> _activeStreams = {};
  final Map<String, int> _connectionCounts = {};
  
  void registerStream(String clientId, String endpoint, StreamSubscription subscription) {
    _activeStreams.putIfAbsent(clientId, () => {}).add(subscription);
    _connectionCounts[endpoint] = (_connectionCounts[endpoint] ?? 0) + 1;
    
    logger.info('Client $clientId subscribed to $endpoint', extra: {
      'endpoint': endpoint,
      'total_connections': _connectionCounts[endpoint],
    });
  }
  
  Future<void> disconnectClient(String clientId) async {
    final streams = _activeStreams.remove(clientId);
    if (streams != null) {
      for (final stream in streams) {
        await stream.cancel();
      }
      
      logger.info('Disconnected client $clientId', extra: {
        'streams_closed': streams.length,
      });
    }
  }
  
  Map<String, int> getConnectionStats() {
    return Map.from(_connectionCounts);
  }
  
  Future<void> shutdown() async {
    for (final streams in _activeStreams.values) {
      for (final stream in streams) {
        await stream.cancel();
      }
    }
    _activeStreams.clear();
    _connectionCounts.clear();
  }
}

Database Integration and ORM

Hypermodern servers typically need robust database integration for persistent storage.

Database Connection Setup

class DatabaseManager {
  late Database _db;
  late ConnectionPool _pool;
  
  Future<void> initialize() async {
    _pool = ConnectionPool(
      host: 'localhost',
      port: 5432,
      database: 'hypermodern_app',
      username: 'app_user',
      password: 'secure_password',
      minConnections: 5,
      maxConnections: 20,
      connectionTimeout: Duration(seconds: 10),
      idleTimeout: Duration(minutes: 10),
    );
    
    _db = Database(_pool);
    
    // Run migrations
    await _runMigrations();
    
    // Set up connection monitoring
    _pool.onConnectionCreated.listen((conn) {
      logger.debug('Database connection created: ${conn.id}');
    });
    
    _pool.onConnectionClosed.listen((conn) {
      logger.debug('Database connection closed: ${conn.id}');
    });
  }
  
  Database get database => _db;
  
  Future<void> _runMigrations() async {
    final migrator = Migrator(_db);
    await migrator.runPendingMigrations();
  }
  
  Future<void> close() async {
    await _pool.close();
  }
}

Repository Pattern

abstract class Repository<T> {
  Future<T?> findById(int id);
  Future<List<T>> findAll();
  Future<T> create(T entity);
  Future<T> update(T entity);
  Future<void> delete(int id);
}

class UserRepository implements Repository<User> {
  final Database db;
  
  UserRepository(this.db);
  
  @override
  Future<User?> findById(int id) async {
    final result = await db.query(
      'SELECT * FROM users WHERE id = ? AND deleted_at IS NULL',
      [id],
    );
    
    return result.isEmpty ? null : User.fromJson(result.first);
  }
  
  @override
  Future<List<User>> findAll() async {
    final result = await db.query(
      'SELECT * FROM users WHERE deleted_at IS NULL ORDER BY created_at DESC',
    );
    
    return result.map((row) => User.fromJson(row)).toList();
  }
  
  Future<User?> findByEmail(String email) async {
    final result = await db.query(
      'SELECT * FROM users WHERE email = ? AND deleted_at IS NULL',
      [email],
    );
    
    return result.isEmpty ? null : User.fromJson(result.first);
  }
  
  Future<List<User>> search({
    String? query,
    UserStatus? status,
    DateTime? createdAfter,
    int limit = 50,
    int offset = 0,
  }) async {
    final queryBuilder = QueryBuilder('users')
        .where('deleted_at IS NULL');
    
    if (query != null && query.isNotEmpty) {
      queryBuilder.where(
        '(username ILIKE ? OR email ILIKE ?)',
        ['%$query%', '%$query%'],
      );
    }
    
    if (status != null) {
      queryBuilder.where('status = ?', [status.toString()]);
    }
    
    if (createdAfter != null) {
      queryBuilder.where('created_at > ?', [createdAfter]);
    }
    
    final query_result = queryBuilder
        .orderBy('created_at DESC')
        .limit(limit)
        .offset(offset)
        .build();
    
    final result = await db.query(query_result.sql, query_result.parameters);
    return result.map((row) => User.fromJson(row)).toList();
  }
  
  @override
  Future<User> create(User user) async {
    final result = await db.query(
      '''
      INSERT INTO users (username, email, password_hash, status, created_at, updated_at)
      VALUES (?, ?, ?, ?, ?, ?)
      RETURNING *
      ''',
      [
        user.username,
        user.email,
        user.passwordHash,
        user.status.toString(),
        user.createdAt,
        user.updatedAt,
      ],
    );
    
    return User.fromJson(result.first);
  }
  
  @override
  Future<User> update(User user) async {
    final result = await db.query(
      '''
      UPDATE users 
      SET username = ?, email = ?, status = ?, updated_at = ?
      WHERE id = ? AND deleted_at IS NULL
      RETURNING *
      ''',
      [
        user.username,
        user.email,
        user.status.toString(),
        DateTime.now(),
        user.id,
      ],
    );
    
    if (result.isEmpty) {
      throw NotFoundException('User not found');
    }
    
    return User.fromJson(result.first);
  }
  
  @override
  Future<void> delete(int id) async {
    // Soft delete
    final result = await db.query(
      '''
      UPDATE users 
      SET deleted_at = ?
      WHERE id = ? AND deleted_at IS NULL
      ''',
      [DateTime.now(), id],
    );
    
    if (result.affectedRows == 0) {
      throw NotFoundException('User not found');
    }
  }
}

Transaction Management

class UserService {
  final UserRepository userRepo;
  final ProfileRepository profileRepo;
  final Database db;
  
  UserService(this.userRepo, this.profileRepo, this.db);
  
  Future<User> createUserWithProfile({
    required String username,
    required String email,
    required String password,
    required UserProfile profile,
  }) async {
    return await db.transaction((tx) async {
      // Create user
      final user = await userRepo.create(User(
        username: username,
        email: email,
        passwordHash: _hashPassword(password),
        status: UserStatus.active,
        createdAt: DateTime.now(),
        updatedAt: DateTime.now(),
      ));
      
      // Create profile
      final userProfile = await profileRepo.create(profile.copyWith(
        userId: user.id,
      ));
      
      // Return user with profile
      return user.copyWith(profile: userProfile);
    });
  }
  
  Future<void> transferUserData(int fromUserId, int toUserId) async {
    await db.transaction((tx) async {
      // Verify both users exist
      final fromUser = await userRepo.findById(fromUserId);
      final toUser = await userRepo.findById(toUserId);
      
      if (fromUser == null || toUser == null) {
        throw NotFoundException('One or both users not found');
      }
      
      // Transfer posts
      await tx.query(
        'UPDATE posts SET author_id = ? WHERE author_id = ?',
        [toUserId, fromUserId],
      );
      
      // Transfer comments
      await tx.query(
        'UPDATE comments SET user_id = ? WHERE user_id = ?',
        [toUserId, fromUserId],
      );
      
      // Merge user preferences
      await _mergeUserPreferences(tx, fromUserId, toUserId);
      
      // Deactivate source user
      await tx.query(
        'UPDATE users SET status = ?, updated_at = ? WHERE id = ?',
        [UserStatus.inactive.toString(), DateTime.now(), fromUserId],
      );
    });
  }
}

Background Job Scheduling

Hypermodern servers include a built-in job scheduling system for handling background tasks, delayed execution, and long-running processes without blocking your main application.

Quick Start with Scheduling

import 'package:hypermodern_server/hypermodern_server.dart';

// Define a scheduled job
class SendWelcomeEmailJob extends ScheduledJob {
  @override
  String get identifier => 'send_welcome_email';

  @override
  Future<void> execute(Map<String, dynamic> parameters) async {
    final email = parameters['email'] as String;
    final name = parameters['name'] as String;
    
    await emailService.sendWelcomeEmail(email, name);
  }
}

void main() async {
  final server = HypermodernServer();
  
  // Set up job scheduling
  final jobRegistry = JobRegistry();
  final scheduler = JobScheduler(registry: jobRegistry);
  
  // Register jobs
  jobRegistry.register('send_welcome_email', () => SendWelcomeEmailJob());
  
  // Start the scheduler
  scheduler.start();
  
  // In your endpoint handlers, schedule jobs
  server.registerEndpoint<CreateUserRequest, User>(
    'create_user',
    (request) async {
      final user = await userService.createUser(request);
      
      // Schedule welcome email for 5 minutes later
      await scheduler.scheduleDelayed(
        'send_welcome_email',
        Duration(minutes: 5),
        {
          'email': user.email,
          'name': user.username,
        },
        description: 'Send welcome email to new user',
      );
      
      return user;
    },
  );
  
  await server.listen();
}

Integration with Server Lifecycle

class HypermodernServerWithScheduling extends HypermodernServer {
  late JobScheduler _scheduler;
  late JobRegistry _jobRegistry;
  
  @override
  Future<void> initialize() async {
    await super.initialize();
    
    // Initialize scheduling system
    _jobRegistry = JobRegistry();
    _scheduler = JobScheduler(
      registry: _jobRegistry,
      executionInterval: Duration(seconds: 30),
      maxTasksPerCycle: 50,
    );
    
    // Register your jobs
    _registerJobs();
    
    // Start scheduler
    _scheduler.start();
  }
  
  void _registerJobs() {
    _jobRegistry.register('send_welcome_email', () => SendWelcomeEmailJob());
    _jobRegistry.register('cleanup_old_data', () => CleanupJob());
    _jobRegistry.register('generate_report', () => ReportJob());
  }
  
  JobScheduler get scheduler => _scheduler;
  
  @override
  Future<void> shutdown() async {
    _scheduler.stop();
    await super.shutdown();
  }
}

The scheduling system provides persistent job storage, automatic retries, error handling, and comprehensive monitoring. For detailed information on creating complex jobs, handling failures, and performance optimization, see Chapter 11.5: Background Job Scheduling.

What's Next

You now have a solid foundation in Hypermodern server development.development, including background job scheduling. In the next chapter, we'll explore the multi-protocol communication system in detail, learning how to optimize for each protocol's strengths while maintaining a unified codebase.

Back to top