Node.js批量发消息接口，Angular实时展示每条结果实现方法

Let's break this down step by step—your current setup has two critical issues blocking real-time updates, and we'll fix both the backend and frontend to make this work smoothly.

First: Fix the Node.js Backend #

Your current code has two big problems:

1. You're using a GET request but trying to read req.body.message—GET requests don't carry a request body by default. Use req.query.message instead, or switch to a POST request (better for sending message content).
2. Calling res.send() inside the forEach loop will throw an error because an HTTP request can only have one response. To send real-time updates for each user, Server-Sent Events (SSE) is the perfect tool—it lets the server push data to the client one by one without waiting for all tasks to finish.

Here's the revised backend code:

app.get('/sendMessageToAllUsers', (req, res) => {
  // Configure SSE response headers
  res.setHeader('Content-Type', 'text/event-stream');
  res.setHeader('Cache-Control', 'no-cache');
  res.setHeader('Connection', 'keep-alive');

  // Get message from query params (since this is a GET request)
  const message = req.query.message;
  if (!message) {
    res.write(`data: ${JSON.stringify({error: 'Message content is required'})}\n\n`);
    res.end();
    return;
  }

  User.find()
    .then(users => {
      let completedTasks = 0;
      const totalUsers = users.length;

      users.forEach(user => {
        bot.telegram.sendMessage(user.user_id, message)
          .then(() => {
            // Push success result to client via SSE
            const successMsg = { message: `Message sent to ${user.username || user.user_id}` };
            res.write(`data: ${JSON.stringify(successMsg)}\n\n`);
            completedTasks++;
            // Close connection when all tasks are done
            if (completedTasks === totalUsers) {
              res.write(`data: ${JSON.stringify({status: 'All messages processed!'})}\n\n`);
              res.end();
            }
          })
          .catch(err => {
            // Push error result to client via SSE
            const errorMsg = { message: `Failed to send to ${user.username || user.user_id}: ${err.message}` };
            res.write(`data: ${JSON.stringify(errorMsg)}\n\n`);
            completedTasks++;
            if (completedTasks === totalUsers) {
              res.end();
            }
          });
      });
    })
    .catch(err => {
      res.write(`data: ${JSON.stringify({error: `Failed to fetch users: ${err.message}`})}\n\n`);
      res.end();
    });
});

Second: Angular Frontend Implementation #

We'll create an SSE service to manage the connection, then use it in a component to update the table in real time.

Step 1: Create an SSE Service #

// sse.service.ts
import { Injectable } from '@angular/core';
import { Observable } from 'rxjs';

@Injectable({ providedIn: 'root' })
export class SseService {
  getServerSentEvents(url: string): Observable<any> {
    return new Observable(observer => {
      const eventSource = new EventSource(url);

      // Handle incoming messages from the server
      eventSource.onmessage = (event) => {
        const data = JSON.parse(event.data);
        observer.next(data);
      };

      // Handle errors or connection closure
      eventSource.onerror = (error) => {
        if (eventSource.readyState === EventSource.CLOSED) {
          observer.complete();
        } else {
          observer.error(error);
        }
      };

      // Cleanup: close connection when component unsubscribes
      return () => {
        eventSource.close();
      };
    });
  }
}

Step 2: Build the Component with Real-Time Table #

// message-batch.component.ts
import { Component } from '@angular/core';
import { SseService } from './sse.service';

@Component({
  selector: 'app-message-batch',
  template: `
    <div class="container">
      <input 
        type="text" 
        [(ngModel)]="messageContent" 
        placeholder="Enter your message"
        class="message-input"
      >
      <button (click)="triggerBatchSend()" class="send-btn">Send to All Users</button>

      <table class="status-table">
        <thead>
          <tr>
            <th>Processing Status</th>
          </tr>
        </thead>
        <tbody>
          <tr *ngFor="let status of statusUpdates">
            <td>{{ status.message }}</td>
          </tr>
        </tbody>
      </table>
    </div>
  `,
  styles: [`
    .message-input { padding: 8px; margin-right: 10px; width: 300px; }
    .send-btn { padding: 8px 16px; cursor: pointer; }
    .status-table { margin-top: 20px; border-collapse: collapse; width: 100%; }
    .status-table th, .status-table td { border: 1px solid #ddd; padding: 8px; text-align: left; }
  `]
})
export class MessageBatchComponent {
  messageContent = '';
  statusUpdates: { message: string }[] = [];

  constructor(private sseService: SseService) {}

  triggerBatchSend() {
    if (!this.messageContent.trim()) return;

    // Clear previous statuses
    this.statusUpdates = [];
    // Encode message to handle special characters in URL
    const encodedMessage = encodeURIComponent(this.messageContent);
    const sseUrl = `/sendMessageToAllUsers?message=${encodedMessage}`;

    // Subscribe to SSE stream
    this.sseService.getServerSentEvents(sseUrl).subscribe({
      next: (data) => {
        if (data.error) {
          this.statusUpdates.push({ message: `❌ ${data.error}` });
        } else {
          this.statusUpdates.push(data);
        }
      },
      complete: () => {
        this.statusUpdates.push({ message: '✅ Batch processing finished!' });
      },
      error: (err) => {
        this.statusUpdates.push({ message: `❌ Connection error: ${err.message}` });
      }
    });
  }
}

Key Notes #

• If you prefer using POST for the request (better for longer messages), you'll need to use the Fetch API with a ReadableStream instead of EventSource (since EventSource only supports GET requests).
• Make sure your Node.js backend has CORS configured correctly if your Angular app runs on a different port/domain.
• The component uses Angular's *ngFor to automatically render new table rows as status updates arrive.

内容的提问来源于stack exchange，提问作者Oshrr Hagag