乐闻世界logo
搜索文章和话题

What is TypeORM Subscriber? How to use subscribers to listen to entity events

2月17日 23:52

Subscribers are a powerful mechanism in TypeORM for listening and responding to entity events. They allow developers to execute custom logic at key moments in an entity's lifecycle, similar to database triggers but more flexible and type-safe.

Subscriber Basic Concepts

What is a Subscriber

A subscriber is a class that listens to lifecycle events of specific entities and executes corresponding logic when these events occur. Events that subscribers can listen to include:

  • beforeInsert: Before insertion
  • afterInsert: After insertion
  • beforeUpdate: Before update
  • afterUpdate: After update
  • beforeRemove: Before deletion
  • afterRemove: After deletion
  • beforeSoftRemove: Before soft deletion
  • afterSoftRemove: After soft deletion
  • beforeRecover: Before recovery
  • afterRecover: After recovery

Subscriber vs Listener

  • Subscriber: Listens to events of all entity instances, suitable for global logic
  • Listener: Defined within the entity, only listens to events of that entity, suitable for entity-specific logic

Creating Subscribers

Basic Subscriber Example

typescript
import { EntitySubscriberInterface, EventSubscriber, InsertEvent, UpdateEvent } from 'typeorm';
import { User } from '../entity/User';

@EventSubscriber()
export class UserSubscriber implements EntitySubscriberInterface<User> {
  // Specify the entity to listen to
  listenTo() {
    return User;
  }

  // Before insertion
  beforeInsert(event: InsertEvent<User>) {
    console.log('Before insert user:', event.entity);
    
    // Auto-generate username
    if (!event.entity.username) {
      event.entity.username = event.entity.email.split('@')[0];
    }
    
    // Auto-set creation time
    if (!event.entity.createdAt) {
      event.entity.createdAt = new Date();
    }
  }

  // After insertion
  afterInsert(event: InsertEvent<User>) {
    console.log('After insert user:', event.entity);
    
    // Send welcome email
    this.sendWelcomeEmail(event.entity);
    
    // Log audit
    this.logAudit('INSERT', event.entity);
  }

  // Before update
  beforeUpdate(event: UpdateEvent<User>) {
    console.log('Before update user:', event.entity);
    
    // Auto-update modification time
    if (event.entity) {
      event.entity.updatedAt = new Date();
    }
  }

  // After update
  afterUpdate(event: UpdateEvent<User>) {
    console.log('After update user:', event.entity);
    
    // Record change history
    this.recordChangeHistory(event);
  }

  // Before deletion
  beforeRemove(event: any) {
    console.log('Before remove user:', event.entity);
    
    // Check if deletion is allowed
    if (event.entity.hasActiveOrders()) {
      throw new Error('Cannot delete user with active orders');
    }
  }

  // After deletion
  afterRemove(event: any) {
    console.log('After remove user:', event.entity);
    
    // Clean up related data
    this.cleanupRelatedData(event.entity.id);
  }

  private sendWelcomeEmail(user: User) {
    // Logic to send welcome email
    console.log(`Sending welcome email to ${user.email}`);
  }

  private logAudit(action: string, user: User) {
    // Logic to log audit
    console.log(`Audit log: ${action} user ${user.id}`);
  }

  private recordChangeHistory(event: UpdateEvent<User>) {
    // Logic to record change history
    console.log('Recording change history:', event.databaseEntity, event.entity);
  }

  private cleanupRelatedData(userId: number) {
    // Logic to clean up related data
    console.log(`Cleaning up data for user ${userId}`);
  }
}

Registering Subscribers

Register in DataSource

typescript
import { DataSource } from 'typeorm';
import { UserSubscriber } from './subscriber/UserSubscriber';

export const AppDataSource = new DataSource({
  type: 'mysql',
  host: 'localhost',
  port: 3306,
  username: 'root',
  password: 'password',
  database: 'myapp',
  entities: [User, Post, Comment],
  subscribers: [UserSubscriber], // Register subscriber
  synchronize: false,
  logging: true,
});

Dynamic Subscriber Registration

typescript
import { DataSource } from 'typeorm';

const dataSource = new DataSource({
  type: 'mysql',
  host: 'localhost',
  port: 3306,
  username: 'root',
  password: 'password',
  database: 'myapp',
  entities: [User, Post, Comment],
  synchronize: false,
  logging: true,
});

// Dynamically add subscriber after initialization
await dataSource.initialize();

const userSubscriber = new UserSubscriber();
dataSource.subscribers.push(userSubscriber);

Advanced Subscriber Usage

Data Validation

typescript
@EventSubscriber()
export class UserSubscriber implements EntitySubscriberInterface<User> {
  listenTo() {
    return User;
  }

  beforeInsert(event: InsertEvent<User>) {
    this.validateUser(event.entity);
  }

  beforeUpdate(event: UpdateEvent<User>) {
    if (event.entity) {
      this.validateUser(event.entity);
    }
  }

  private validateUser(user: User) {
    // Validate email format
    const emailRegex = /^[^\s@]+@[^\s@]+\.[^\s@]+$/;
    if (!emailRegex.test(user.email)) {
      throw new Error('Invalid email format');
    }

    // Validate age
    if (user.age && (user.age < 18 || user.age > 120)) {
      throw new Error('Age must be between 18 and 120');
    }

    // Validate username length
    if (user.username && user.username.length < 3) {
      throw new Error('Username must be at least 3 characters');
    }
  }
}

Auto-fill Fields

typescript
@EventSubscriber()
export class BaseEntitySubscriber implements EntitySubscriberInterface {
  listenTo() {
    return Object; // Listen to all entities
  }

  beforeInsert(event: InsertEvent<any>) {
    const entity = event.entity;
    const now = new Date();

    // Auto-set creation time
    if (entity.hasOwnProperty('createdAt') && !entity.createdAt) {
      entity.createdAt = now;
    }

    // Auto-set update time
    if (entity.hasOwnProperty('updatedAt') && !entity.updatedAt) {
      entity.updatedAt = now;
    }

    // Auto-set creator
    if (entity.hasOwnProperty('createdBy') && !entity.createdBy) {
      entity.createdBy = this.getCurrentUserId();
    }
  }

  beforeUpdate(event: UpdateEvent<any>) {
    const entity = event.entity;
    if (entity) {
      // Auto-update update time
      if (entity.hasOwnProperty('updatedAt')) {
        entity.updatedAt = new Date();
      }

      // Auto-set updater
      if (entity.hasOwnProperty('updatedBy')) {
        entity.updatedBy = this.getCurrentUserId();
      }
    }
  }

  private getCurrentUserId(): number {
    // Logic to get current user ID
    return 1; // Example
  }
}

Audit Logging

typescript
@EventSubscriber()
export class AuditLogSubscriber implements EntitySubscriberInterface {
  listenTo() {
    return Object; // Listen to all entities
  }

  async afterInsert(event: InsertEvent<any>) {
    await this.createAuditLog('INSERT', event.entity);
  }

  async afterUpdate(event: UpdateEvent<any>) {
    await this.createAuditLog('UPDATE', event.entity, event.databaseEntity);
  }

  async afterRemove(event: any) {
    await this.createAuditLog('DELETE', event.entity);
  }

  private async createAuditLog(
    action: string,
    entity: any,
    oldEntity?: any
  ) {
    const auditLog = {
      action,
      entityType: entity.constructor.name,
      entityId: entity.id,
      userId: this.getCurrentUserId(),
      timestamp: new Date(),
      changes: oldEntity ? this.getChanges(oldEntity, entity) : null,
      ipAddress: this.getCurrentIpAddress(),
    };

    // Save audit log
    console.log('Creating audit log:', auditLog);
    // await this.auditLogRepository.save(auditLog);
  }

  private getChanges(oldEntity: any, newEntity: any): any {
    const changes: any = {};
    
    for (const key in newEntity) {
      if (oldEntity[key] !== newEntity[key]) {
        changes[key] = {
          old: oldEntity[key],
          new: newEntity[key],
        };
      }
    }
    
    return changes;
  }

  private getCurrentUserId(): number {
    return 1; // Example
  }

  private getCurrentIpAddress(): string {
    return '127.0.0.1'; // Example
  }
}

Cache Invalidation

typescript
@EventSubscriber()
export class CacheInvalidationSubscriber implements EntitySubscriberInterface {
  private cacheService: CacheService;

  constructor() {
    this.cacheService = new CacheService();
  }

  listenTo() {
    return Object; // Listen to all entities
  }

  async afterInsert(event: InsertEvent<any>) {
    await this.invalidateCache(event.entity);
  }

  async afterUpdate(event: UpdateEvent<any>) {
    if (event.entity) {
      await this.invalidateCache(event.entity);
    }
  }

  async afterRemove(event: any) {
    await this.invalidateCache(event.entity);
  }

  private async invalidateCache(entity: any) {
    const entityType = entity.constructor.name.toLowerCase();
    const entityId = entity.id;

    // Invalidate single entity cache
    await this.cacheService.delete(`${entityType}:${entityId}`);

    // Invalidate list cache
    await this.cacheService.delete(`${entityType}:list:*`);

    console.log(`Cache invalidated for ${entityType}:${entityId}`);
  }
}

Notifications and Events

typescript
@EventSubscriber()
export class NotificationSubscriber implements EntitySubscriberInterface {
  private notificationService: NotificationService;

  constructor() {
    this.notificationService = new NotificationService();
  }

  listenTo() {
    return Object; // Listen to all entities
  }

  async afterInsert(event: InsertEvent<any>) {
    await this.handleInsertEvent(event);
  }

  async afterUpdate(event: UpdateEvent<any>) {
    await this.handleUpdateEvent(event);
  }

  private async handleInsertEvent(event: InsertEvent<any>) {
    const entity = event.entity;

    // Send different notifications based on entity type
    switch (entity.constructor.name) {
      case 'Order':
        await this.notificationService.sendOrderCreatedNotification(entity);
        break;
      case 'Comment':
        await this.notificationService.sendCommentNotification(entity);
        break;
      case 'Message':
        await this.notificationService.sendMessageNotification(entity);
        break;
    }
  }

  private async handleUpdateEvent(event: UpdateEvent<any>) {
    const entity = event.entity;
    if (!entity) return;

    // Send notifications based on entity type and changes
    switch (entity.constructor.name) {
      case 'Order':
        if (entity.status !== event.databaseEntity.status) {
          await this.notificationService.sendOrderStatusChangedNotification(entity);
        }
        break;
      case 'User':
        if (entity.email !== event.databaseEntity.email) {
          await this.notificationService.sendEmailChangedNotification(entity);
        }
        break;
    }
  }
}

Subscriber Best Practices

1. Single Responsibility Principle

Each subscriber should be responsible for only one specific functional area.

typescript
// ✅ Good: Each subscriber handles one function
@EventSubscriber()
export class UserValidationSubscriber implements EntitySubscriberInterface<User> {
  listenTo() { return User; }
  beforeInsert(event: InsertEvent<User>) { /* Validation logic */ }
}

@EventSubscriber()
export class UserAuditSubscriber implements EntitySubscriberInterface<User> {
  listenTo() { return User; }
  afterInsert(event: InsertEvent<User>) { /* Audit logic */ }
}

// ❌ Bad: One subscriber handles multiple functions
@EventSubscriber()
export class UserSubscriber implements EntitySubscriberInterface<User> {
  listenTo() { return User; }
  beforeInsert(event: InsertEvent<User>) { /* Validation logic */ }
  afterInsert(event: InsertEvent<User>) { /* Audit logic */ }
  afterUpdate(event: UpdateEvent<User>) { /* Notification logic */ }
}

2. Avoid Circular Dependencies

Subscribers should not trigger operations that cause infinite loops in other subscribers.

typescript
@EventSubscriber()
export class UserSubscriber implements EntitySubscriberInterface<User> {
  constructor(
    private userRepository: Repository<User>
  ) {}

  listenTo() { return User; }

  async afterInsert(event: InsertEvent<User>) {
    // ❌ Bad: May cause circular dependency
    // await this.userRepository.save(event.entity);

    // ✅ Good: Use EntityManager to avoid triggering subscribers
    await this.userRepository.manager.save(User, event.entity);
  }
}

3. Error Handling

Handle errors properly in subscribers to avoid affecting main operations.

typescript
@EventSubscriber()
export class SafeSubscriber implements EntitySubscriberInterface<User> {
  listenTo() { return User; }

  async afterInsert(event: InsertEvent<User>) {
    try {
      await this.sendNotification(event.entity);
    } catch (error) {
      // Log error but don't affect main operation
      console.error('Failed to send notification:', error);
      // Can send error to error monitoring system
    }
  }

  private async sendNotification(user: User) {
    // Logic to send notification
  }
}

4. Performance Considerations

Avoid executing time-consuming operations in subscribers.

typescript
@EventSubscriber()
export class PerformanceAwareSubscriber implements EntitySubscriberInterface<User> {
  listenTo() { return User; }

  async afterInsert(event: InsertEvent<User>) {
    // ❌ Bad: Synchronous execution of time-consuming operations
    // await this.sendEmail(event.entity);
    // await this.generateReport(event.entity);

    // ✅ Good: Asynchronous execution of time-consuming operations
    setImmediate(() => {
      this.sendEmail(event.entity).catch(console.error);
      this.generateReport(event.entity).catch(console.error);
    });

    // Or use message queue
    // await this.queueService.add('send-email', { userId: event.entity.id });
  }

  private async sendEmail(user: User) {
    // Logic to send email
  }

  private async generateReport(user: User) {
    // Logic to generate report
  }
}

5. Testing Subscribers

Write unit tests for subscribers.

typescript
import { InsertEvent } from 'typeorm';
import { UserSubscriber } from './UserSubscriber';
import { User } from '../entity/User';

describe('UserSubscriber', () => {
  let subscriber: UserSubscriber;

  beforeEach(() => {
    subscriber = new UserSubscriber();
  });

  it('should auto-generate username before insert', () => {
    const user = new User();
    user.email = 'test@example.com';

    const event: InsertEvent<User> = {
      entity: user,
      metadata: {} as any,
      queryRunner: {} as any,
      manager: {} as any,
    };

    subscriber.beforeInsert(event);

    expect(user.username).toBe('test');
  });

  it('should set createdAt before insert', () => {
    const user = new User();
    user.email = 'test@example.com';

    const event: InsertEvent<User> = {
      entity: user,
      metadata: {} as any,
      queryRunner: {} as any,
      manager: {} as any,
    };

    subscriber.beforeInsert(event);

    expect(user.createdAt).toBeInstanceOf(Date);
  });
});

Subscriber vs Listener

Listener Example

typescript
@Entity()
export class User {
  @PrimaryGeneratedColumn()
  id: number;

  @Column()
  name: string;

  @Column()
  email: string;

  @CreateDateColumn()
  createdAt: Date;

  @UpdateDateColumn()
  updatedAt: Date;

  @BeforeInsert()
  beforeInsert() {
    console.log('Before insert in entity');
    this.createdAt = new Date();
  }

  @AfterInsert()
  afterInsert() {
    console.log('After insert in entity');
  }

  @BeforeUpdate()
  beforeUpdate() {
    console.log('Before update in entity');
    this.updatedAt = new Date();
  }

  @AfterUpdate()
  afterUpdate() {
    console.log('After update in entity');
  }
}

Choosing Subscriber or Listener

Use Subscriber when:

  • Need to listen to events of multiple entities
  • Need to inject other services
  • Need to implement cross-entity business logic
  • Need to keep entity classes clean

Use Listener when:

  • Logic is only related to a single entity
  • Logic is simple and doesn't need external dependencies
  • Want logic to be tightly coupled with the entity

TypeORM's subscriber mechanism provides powerful and flexible event handling capabilities. Proper use of subscribers can implement complex business logic while keeping code clean and maintainable.

标签：TypeORM
热门标签
更多
Git(114)C语言(23)C++(15)前端(148)React(32)JavaScript(56)Linux(20)Rust(10)Docker(65)Cypress(4)TypeScript(23)Tailwind CSS(29)Vue(13)CSS(19)网络(12)MySQL(7)TypeORM(31)Next.js(30)