Feathers.js 实战项目：聊天应用 - 完整的实时通信系统

发布时间：2024-07-05
作者：一介布衣
标签：Feathers.js, 聊天应用, 实时通信, WebSocket, 项目实战

前言

经过前面十几篇文章的学习，相信大家对 Feathers.js 已经有了深入的了解。今天咱们来做一个完整的实战项目 - 构建一个功能丰富的聊天应用。说实话，聊天应用是学习实时通信最好的项目，它涵盖了用户管理、消息处理、文件上传、实时同步等各个方面。

我记得第一次做聊天应用的时候，觉得很简单：不就是发消息、收消息嘛。后来才发现，要做一个好用的聊天应用，需要考虑的东西太多了：消息状态、离线消息、文件传输、表情包、群聊管理、消息搜索等等。用 Feathers.js 来实现，正好可以体验它在实时应用方面的强大能力。

今天我就带大家从零开始，构建一个功能完整的聊天应用。

项目需求分析

核心功能

用户系统：

• 用户注册和登录
• 个人资料管理
• 在线状态显示
• 用户搜索和添加好友

聊天功能：

• 一对一私聊
• 群聊创建和管理
• 消息发送和接收
• 消息状态（已发送、已送达、已读）
• 离线消息处理

消息类型：

• 文本消息
• 图片和文件
• 表情包
• 语音消息
• 位置分享

高级功能：

• 消息搜索
• 聊天记录导出
• 消息撤回
• @提醒功能
• 消息加密

数据模型设计

1. 用户模型

// src/models/users.model.js
module.exports = function (app) {
  const mongoClient = app.get('mongoClient');
  
  mongoClient.then(db => {
    const collection = db.collection('users');
    
    collection.createIndex({ email: 1 }, { unique: true });
    collection.createIndex({ username: 1 }, { unique: true });
    collection.createIndex({ 'status.isOnline': 1 });
  });

  return mongoClient;
};

// 用户文档结构
const userSchema = {
  _id: ObjectId,
  email: String,
  username: String,
  password: String,  // 加密后的密码
  
  profile: {
    displayName: String,
    avatar: String,
    bio: String,
    phone: String
  },
  
  status: {
    isOnline: Boolean,
    lastSeen: Date,
    currentActivity: String  // 'typing', 'idle', 'away'
  },
  
  settings: {
    notifications: {
      sound: Boolean,
      desktop: Boolean,
      mobile: Boolean
    },
    privacy: {
      showLastSeen: Boolean,
      showOnlineStatus: Boolean,
      allowStrangerMessages: Boolean
    },
    theme: String  // 'light', 'dark', 'auto'
  },
  
  contacts: [{
    userId: ObjectId,
    addedAt: Date,
    nickname: String,
    isBlocked: Boolean
  }],
  
  createdAt: Date,
  updatedAt: Date
};

2. 聊天室模型

// src/models/rooms.model.js
const roomSchema = {
  _id: ObjectId,
  name: String,  // 群聊名称，私聊为空
  description: String,
  avatar: String,
  
  type: String,  // 'private', 'group', 'channel'
  
  // 成员管理
  members: [{
    userId: ObjectId,
    role: String,  // 'owner', 'admin', 'member'
    joinedAt: Date,
    nickname: String,
    permissions: [String]  // 'send_messages', 'add_members', 'delete_messages'
  }],
  
  // 聊天室设置
  settings: {
    isPublic: Boolean,
    allowInvites: Boolean,
    maxMembers: Number,
    messageRetention: Number,  // 消息保留天数
    allowFileUpload: Boolean,
    allowVoiceMessages: Boolean
  },
  
  // 统计信息
  stats: {
    messageCount: Number,
    memberCount: Number,
    lastActivity: Date
  },
  
  // 最后一条消息
  lastMessage: {
    messageId: ObjectId,
    content: String,
    senderId: ObjectId,
    timestamp: Date,
    type: String
  },
  
  createdAt: Date,
  updatedAt: Date,
  createdBy: ObjectId
};

3. 消息模型

// src/models/messages.model.js
const messageSchema = {
  _id: ObjectId,
  roomId: ObjectId,
  senderId: ObjectId,
  
  // 消息内容
  content: String,
  type: String,  // 'text', 'image', 'file', 'voice', 'location', 'system'
  
  // 媒体文件信息
  media: {
    url: String,
    filename: String,
    size: Number,
    mimeType: String,
    duration: Number,  // 语音消息时长
    thumbnail: String  // 图片缩略图
  },
  
  // 位置信息
  location: {
    latitude: Number,
    longitude: Number,
    address: String
  },
  
  // 回复消息
  replyTo: {
    messageId: ObjectId,
    content: String,
    senderId: ObjectId
  },
  
  // 转发信息
  forwarded: {
    originalMessageId: ObjectId,
    originalSenderId: ObjectId,
    forwardCount: Number
  },
  
  // 消息状态
  status: {
    sent: Date,
    delivered: [{ userId: ObjectId, timestamp: Date }],
    read: [{ userId: ObjectId, timestamp: Date }]
  },
  
  // 编辑历史
  editHistory: [{
    content: String,
    editedAt: Date
  }],
  
  // 反应表情
  reactions: [{
    userId: ObjectId,
    emoji: String,
    timestamp: Date
  }],
  
  // 提醒用户
  mentions: [ObjectId],
  
  // 系统标记
  isDeleted: Boolean,
  deletedAt: Date,
  deletedBy: ObjectId,
  
  createdAt: Date,
  updatedAt: Date
};

核心服务实现

1. 用户服务

// src/services/users/users.class.js
const { MongoDBService } = require('@feathersjs/mongodb');
const { ObjectId } = require('mongodb');

class UsersService extends MongoDBService {
  constructor(options, app) {
    super(options, app);
    this.app = app;
  }

  async create(data, params) {
    const userData = {
      ...data,
      status: {
        isOnline: false,
        lastSeen: new Date(),
        currentActivity: 'offline'
      },
      settings: {
        notifications: {
          sound: true,
          desktop: true,
          mobile: true
        },
        privacy: {
          showLastSeen: true,
          showOnlineStatus: true,
          allowStrangerMessages: true
        },
        theme: 'light'
      },
      contacts: [],
      createdAt: new Date(),
      updatedAt: new Date()
    };

    return super.create(userData, params);
  }

  // 更新在线状态
  async updateOnlineStatus(userId, isOnline, activity = 'online') {
    const updateData = {
      'status.isOnline': isOnline,
      'status.lastSeen': new Date(),
      'status.currentActivity': activity,
      updatedAt: new Date()
    };

    await super.patch(userId, updateData);

    // 广播状态变化给联系人
    const user = await this.get(userId);
    const contactIds = user.contacts.map(contact => contact.userId);

    this.app.channel('authenticated').send({
      type: 'user-status-changed',
      userId: userId,
      isOnline: isOnline,
      activity: activity,
      lastSeen: updateData['status.lastSeen']
    });

    return { success: true };
  }

  // 添加联系人
  async addContact(userId, contactId, params) {
    // 检查是否已经是联系人
    const user = await this.get(userId);
    const existingContact = user.contacts.find(
      contact => contact.userId.toString() === contactId
    );

    if (existingContact) {
      throw new Error('用户已经在联系人列表中');
    }

    // 添加联系人
    await super.patch(userId, {
      $push: {
        contacts: {
          userId: new ObjectId(contactId),
          addedAt: new Date(),
          nickname: '',
          isBlocked: false
        }
      }
    });

    // 创建私聊房间
    const room = await this.app.service('rooms').create({
      type: 'private',
      members: [
        {
          userId: new ObjectId(userId),
          role: 'member',
          joinedAt: new Date()
        },
        {
          userId: new ObjectId(contactId),
          role: 'member',
          joinedAt: new Date()
        }
      ],
      settings: {
        isPublic: false,
        allowInvites: false,
        maxMembers: 2
      }
    });

    return { success: true, roomId: room._id };
  }

  // 搜索用户
  async searchUsers(query, params) {
    const { user } = params;
    
    const pipeline = [
      {
        $match: {
          $and: [
            { _id: { $ne: new ObjectId(user._id) } },
            {
              $or: [
                { username: { $regex: query, $options: 'i' } },
                { 'profile.displayName': { $regex: query, $options: 'i' } },
                { email: { $regex: query, $options: 'i' } }
              ]
            }
          ]
        }
      },
      {
        $project: {
          username: 1,
          'profile.displayName': 1,
          'profile.avatar': 1,
          'status.isOnline': 1,
          'status.lastSeen': 1
        }
      },
      { $limit: 20 }
    ];

    const users = await this.Model.aggregate(pipeline).toArray();
    
    return {
      total: users.length,
      data: users
    };
  }

  // 获取联系人列表
  async getContacts(userId, params) {
    const user = await this.get(userId);
    const contactIds = user.contacts
      .filter(contact => !contact.isBlocked)
      .map(contact => contact.userId);

    if (contactIds.length === 0) {
      return { total: 0, data: [] };
    }

    const contacts = await this.find({
      query: {
        _id: { $in: contactIds },
        $select: [
          'username',
          'profile.displayName',
          'profile.avatar',
          'status.isOnline',
          'status.lastSeen',
          'status.currentActivity'
        ]
      }
    });

    return contacts;
  }
}

module.exports = UsersService;

2. 聊天室服务

// src/services/rooms/rooms.class.js
const { MongoDBService } = require('@feathersjs/mongodb');
const { ObjectId } = require('mongodb');

class RoomsService extends MongoDBService {
  constructor(options, app) {
    super(options, app);
    this.app = app;
  }

  async find(params) {
    const { user } = params;
    
    // 只返回用户参与的聊天室
    const pipeline = [
      {
        $match: {
          'members.userId': new ObjectId(user._id)
        }
      },
      {
        $addFields: {
          // 添加用户在此房间的角色
          userRole: {
            $arrayElemAt: [
              {
                $filter: {
                  input: '$members',
                  cond: { $eq: ['$$this.userId', new ObjectId(user._id)] }
                }
              },
              0
            ]
          }
        }
      },
      {
        $lookup: {
          from: 'users',
          localField: 'members.userId',
          foreignField: '_id',
          as: 'memberDetails',
          pipeline: [
            {
              $project: {
                username: 1,
                'profile.displayName': 1,
                'profile.avatar': 1,
                'status.isOnline': 1
              }
            }
          ]
        }
      },
      {
        $addFields: {
          // 为私聊生成显示名称
          displayName: {
            $cond: {
              if: { $eq: ['$type', 'private'] },
              then: {
                $let: {
                  vars: {
                    otherMember: {
                      $arrayElemAt: [
                        {
                          $filter: {
                            input: '$memberDetails',
                            cond: { $ne: ['$$this._id', new ObjectId(user._id)] }
                          }
                        },
                        0
                      ]
                    }
                  },
                  in: {
                    $ifNull: [
                      '$$otherMember.profile.displayName',
                      '$$otherMember.username'
                    ]
                  }
                }
              },
              else: '$name'
            }
          },
          // 为私聊生成头像
          displayAvatar: {
            $cond: {
              if: { $eq: ['$type', 'private'] },
              then: {
                $let: {
                  vars: {
                    otherMember: {
                      $arrayElemAt: [
                        {
                          $filter: {
                            input: '$memberDetails',
                            cond: { $ne: ['$$this._id', new ObjectId(user._id)] }
                          }
                        },
                        0
                      ]
                    }
                  },
                  in: '$$otherMember.profile.avatar'
                }
              },
              else: '$avatar'
            }
          }
        }
      },
      {
        $sort: { 'stats.lastActivity': -1 }
      },
      {
        $project: {
          memberDetails: 0  // 清理临时字段
        }
      }
    ];

    const rooms = await this.Model.aggregate(pipeline).toArray();
    
    return {
      total: rooms.length,
      data: rooms
    };
  }

  async create(data, params) {
    const { user } = params;
    
    const roomData = {
      ...data,
      stats: {
        messageCount: 0,
        memberCount: data.members ? data.members.length : 0,
        lastActivity: new Date()
      },
      createdAt: new Date(),
      updatedAt: new Date(),
      createdBy: new ObjectId(user._id)
    };

    const room = await super.create(roomData, params);

    // 让所有成员加入实时频道
    room.members.forEach(member => {
      this.app.channel(`user-${member.userId}`).join(
        this.app.channel(`room-${room._id}`)
      );
    });

    return room;
  }

  // 添加成员
  async addMember(roomId, userId, params) {
    const { user } = params;
    const room = await this.get(roomId);

    // 检查权限
    const userMember = room.members.find(
      member => member.userId.toString() === user._id
    );

    if (!userMember || !['owner', 'admin'].includes(userMember.role)) {
      throw new Error('无权限添加成员');
    }

    // 检查是否已经是成员
    const existingMember = room.members.find(
      member => member.userId.toString() === userId
    );

    if (existingMember) {
      throw new Error('用户已经是群成员');
    }

    // 检查群成员上限
    if (room.members.length >= room.settings.maxMembers) {
      throw new Error('群成员已达上限');
    }

    // 添加成员
    await super.patch(roomId, {
      $push: {
        members: {
          userId: new ObjectId(userId),
          role: 'member',
          joinedAt: new Date(),
          permissions: ['send_messages']
        }
      },
      $inc: { 'stats.memberCount': 1 },
      updatedAt: new Date()
    });

    // 发送系统消息
    await this.app.service('messages').create({
      roomId: new ObjectId(roomId),
      senderId: new ObjectId(user._id),
      type: 'system',
      content: `${user.profile?.displayName || user.username} 邀请了新成员加入群聊`,
      createdAt: new Date()
    });

    // 让新成员加入实时频道
    this.app.channel(`user-${userId}`).join(
      this.app.channel(`room-${roomId}`)
    );

    return { success: true };
  }

  // 离开群聊
  async leaveRoom(roomId, params) {
    const { user } = params;
    const room = await this.get(roomId);

    const memberIndex = room.members.findIndex(
      member => member.userId.toString() === user._id
    );

    if (memberIndex === -1) {
      throw new Error('您不是群成员');
    }

    const member = room.members[memberIndex];

    // 如果是群主，需要转让群主权限
    if (member.role === 'owner' && room.members.length > 1) {
      // 找到下一个管理员或成员作为新群主
      const newOwner = room.members.find(m => 
        m.userId.toString() !== user._id && m.role === 'admin'
      ) || room.members.find(m => 
        m.userId.toString() !== user._id
      );

      if (newOwner) {
        await super.patch(roomId, {
          $set: {
            'members.$[owner].role': 'owner'
          }
        }, {
          arrayFilters: [{ 'owner.userId': newOwner.userId }]
        });
      }
    }

    // 移除成员
    await super.patch(roomId, {
      $pull: { members: { userId: new ObjectId(user._id) } },
      $inc: { 'stats.memberCount': -1 },
      updatedAt: new Date()
    });

    // 如果是最后一个成员，删除群聊
    if (room.members.length === 1) {
      await super.remove(roomId);
    } else {
      // 发送系统消息
      await this.app.service('messages').create({
        roomId: new ObjectId(roomId),
        senderId: new ObjectId(user._id),
        type: 'system',
        content: `${user.profile?.displayName || user.username} 离开了群聊`,
        createdAt: new Date()
      });
    }

    // 离开实时频道
    this.app.channel(`user-${user._id}`).leave(
      this.app.channel(`room-${roomId}`)
    );

    return { success: true };
  }

  // 更新最后消息
  async updateLastMessage(roomId, message) {
    await super.patch(roomId, {
      lastMessage: {
        messageId: message._id,
        content: message.content,
        senderId: message.senderId,
        timestamp: message.createdAt,
        type: message.type
      },
      'stats.lastActivity': new Date(),
      updatedAt: new Date()
    });
  }
}

module.exports = RoomsService;

3. 消息服务

// src/services/messages/messages.class.js
const { MongoDBService } = require('@feathersjs/mongodb');
const { ObjectId } = require('mongodb');

class MessagesService extends MongoDBService {
  constructor(options, app) {
    super(options, app);
    this.app = app;
  }

  async find(params) {
    const { query = {} } = params;
    const { roomId } = query;

    if (!roomId) {
      throw new Error('roomId 是必需的');
    }

    // 验证用户是否有权限访问此聊天室
    const room = await this.app.service('rooms').get(roomId);
    const isMember = room.members.some(
      member => member.userId.toString() === params.user._id
    );

    if (!isMember) {
      throw new Error('无权限访问此聊天室');
    }

    const pipeline = [
      {
        $match: {
          roomId: new ObjectId(roomId),
          isDeleted: { $ne: true }
        }
      },
      {
        $lookup: {
          from: 'users',
          localField: 'senderId',
          foreignField: '_id',
          as: 'sender',
          pipeline: [
            {
              $project: {
                username: 1,
                'profile.displayName': 1,
                'profile.avatar': 1
              }
            }
          ]
        }
      },
      {
        $addFields: {
          sender: { $arrayElemAt: ['$sender', 0] }
        }
      },
      {
        $sort: { createdAt: -1 }
      }
    ];

    // 分页
    if (query.$skip) {
      pipeline.push({ $skip: parseInt(query.$skip) });
    }

    if (query.$limit) {
      pipeline.push({ $limit: parseInt(query.$limit) });
    }

    const messages = await this.Model.aggregate(pipeline).toArray();

    return {
      total: messages.length,
      data: messages.reverse()  // 按时间正序返回
    };
  }

  async create(data, params) {
    const { user } = params;
    
    // 验证用户是否有权限发送消息
    const room = await this.app.service('rooms').get(data.roomId);
    const member = room.members.find(
      m => m.userId.toString() === user._id
    );

    if (!member) {
      throw new Error('您不是此聊天室的成员');
    }

    if (!member.permissions.includes('send_messages')) {
      throw new Error('您没有发送消息的权限');
    }

    const messageData = {
      ...data,
      senderId: new ObjectId(user._id),
      roomId: new ObjectId(data.roomId),
      status: {
        sent: new Date(),
        delivered: [],
        read: []
      },
      reactions: [],
      editHistory: [],
      isDeleted: false,
      createdAt: new Date(),
      updatedAt: new Date()
    };

    // 处理回复消息
    if (data.replyTo) {
      const originalMessage = await this.get(data.replyTo.messageId);
      messageData.replyTo = {
        messageId: new ObjectId(data.replyTo.messageId),
        content: originalMessage.content.substring(0, 100),
        senderId: originalMessage.senderId
      };
    }

    // 处理提醒
    if (data.mentions && data.mentions.length > 0) {
      messageData.mentions = data.mentions.map(id => new ObjectId(id));
    }

    const message = await super.create(messageData, params);

    // 更新聊天室统计
    await this.app.service('rooms').patch(data.roomId, {
      $inc: { 'stats.messageCount': 1 }
    });

    // 更新聊天室最后消息
    await this.app.service('rooms').updateLastMessage(data.roomId, message);

    // 标记消息为已送达（给在线用户）
    await this.markAsDelivered(message._id, room.members);

    return message;
  }

  async patch(id, data, params) {
    const { user } = params;
    const message = await this.get(id);

    // 只有发送者可以编辑消息
    if (message.senderId.toString() !== user._id) {
      throw new Error('只能编辑自己的消息');
    }

    // 检查编辑时间限制（15分钟内）
    const editTimeLimit = 15 * 60 * 1000;
    if (Date.now() - new Date(message.createdAt).getTime() > editTimeLimit) {
      throw new Error('消息编辑时间已过期');
    }

    const updateData = {
      ...data,
      updatedAt: new Date()
    };

    // 保存编辑历史
    if (data.content && data.content !== message.content) {
      updateData.$push = {
        editHistory: {
          content: message.content,
          editedAt: new Date()
        }
      };
    }

    return super.patch(id, updateData, params);
  }

  async remove(id, params) {
    const { user } = params;
    const message = await this.get(id);

    // 检查权限
    const room = await this.app.service('rooms').get(message.roomId);
    const member = room.members.find(
      m => m.userId.toString() === user._id
    );

    const canDelete = message.senderId.toString() === user._id ||
                     ['owner', 'admin'].includes(member?.role);

    if (!canDelete) {
      throw new Error('无权限删除此消息');
    }

    // 软删除
    await super.patch(id, {
      isDeleted: true,
      deletedAt: new Date(),
      deletedBy: new ObjectId(user._id)
    });

    return { success: true };
  }

  // 标记消息为已读
  async markAsRead(messageIds, params) {
    const { user } = params;
    
    if (!Array.isArray(messageIds)) {
      messageIds = [messageIds];
    }

    const readTimestamp = new Date();

    await this.Model.updateMany(
      {
        _id: { $in: messageIds.map(id => new ObjectId(id)) },
        'status.read.userId': { $ne: new ObjectId(user._id) }
      },
      {
        $push: {
          'status.read': {
            userId: new ObjectId(user._id),
            timestamp: readTimestamp
          }
        }
      }
    );

    return { success: true, readAt: readTimestamp };
  }

  // 标记消息为已送达
  async markAsDelivered(messageId, members) {
    const onlineMembers = await this.getOnlineMembers(members);
    
    if (onlineMembers.length === 0) return;

    const deliveredUpdates = onlineMembers.map(member => ({
      userId: member.userId,
      timestamp: new Date()
    }));

    await super.patch(messageId, {
      $push: {
        'status.delivered': { $each: deliveredUpdates }
      }
    });
  }

  // 获取在线成员
  async getOnlineMembers(members) {
    const memberIds = members.map(m => m.userId);
    
    const onlineUsers = await this.app.service('users').find({
      query: {
        _id: { $in: memberIds },
        'status.isOnline': true,
        $select: ['_id']
      }
    });

    return onlineUsers.data.map(user => ({ userId: user._id }));
  }

  // 添加反应
  async addReaction(messageId, emoji, params) {
    const { user } = params;
    
    // 检查是否已经有相同反应
    const message = await this.get(messageId);
    const existingReaction = message.reactions.find(
      r => r.userId.toString() === user._id && r.emoji === emoji
    );

    if (existingReaction) {
      // 移除反应
      await super.patch(messageId, {
        $pull: {
          reactions: {
            userId: new ObjectId(user._id),
            emoji: emoji
          }
        }
      });
    } else {
      // 添加反应
      await super.patch(messageId, {
        $push: {
          reactions: {
            userId: new ObjectId(user._id),
            emoji: emoji,
            timestamp: new Date()
          }
        }
      });
    }

    return { success: true };
  }

  // 搜索消息
  async searchMessages(query, roomId, params) {
    const { user } = params;
    
    // 验证权限
    const room = await this.app.service('rooms').get(roomId);
    const isMember = room.members.some(
      member => member.userId.toString() === user._id
    );

    if (!isMember) {
      throw new Error('无权限搜索此聊天室消息');
    }

    const pipeline = [
      {
        $match: {
          roomId: new ObjectId(roomId),
          isDeleted: { $ne: true },
          $text: { $search: query }
        }
      },
      {
        $lookup: {
          from: 'users',
          localField: 'senderId',
          foreignField: '_id',
          as: 'sender',
          pipeline: [
            {
              $project: {
                username: 1,
                'profile.displayName': 1,
                'profile.avatar': 1
              }
            }
          ]
        }
      },
      {
        $addFields: {
          sender: { $arrayElemAt: ['$sender', 0] },
          score: { $meta: 'textScore' }
        }
      },
      {
        $sort: { score: { $meta: 'textScore' } }
      },
      { $limit: 50 }
    ];

    const messages = await this.Model.aggregate(pipeline).toArray();

    return {
      total: messages.length,
      data: messages
    };
  }
}

module.exports = MessagesService;

实时功能配置

1. 频道管理

// src/channels.js
module.exports = function(app) {
  if(typeof app.channel !== 'function') {
    return;
  }

  app.on('connection', connection => {
    app.channel('anonymous').join(connection);
  });

  app.on('login', (authResult, { connection }) => {
    const { user } = authResult;

    if(connection) {
      app.channel('anonymous').leave(connection);
      app.channel('authenticated').join(connection);
      app.channel(`user-${user._id}`).join(connection);

      // 更新用户在线状态
      app.service('users').updateOnlineStatus(user._id, true, 'online');

      // 加入用户参与的聊天室频道
      app.service('rooms').find({ user }).then(rooms => {
        rooms.data.forEach(room => {
          app.channel(`room-${room._id}`).join(connection);
        });
      });
    }
  });

  app.on('disconnect', connection => {
    if (connection.user) {
      // 更新用户离线状态
      app.service('users').updateOnlineStatus(
        connection.user._id, 
        false, 
        'offline'
      );
    }
  });

  // 消息事件发布
  app.service('messages').publish('created', (data, hook) => {
    return app.channel(`room-${data.roomId}`);
  });

  app.service('messages').publish('patched', (data, hook) => {
    return app.channel(`room-${data.roomId}`);
  });

  // 用户状态事件发布
  app.service('users').publish('patched', (data, hook) => {
    if (hook.data['status.isOnline'] !== undefined) {
      return app.channel('authenticated');
    }
    return [];
  });

  // 聊天室事件发布
  app.service('rooms').publish((data, hook) => {
    return app.channel(`room-${data._id}`);
  });
};

前端集成示例

1. 聊天组件

// client/components/ChatRoom.js
import React, { useState, useEffect, useRef } from 'react';
import client from '../feathers-client';

const ChatRoom = ({ roomId, currentUser }) => {
  const [messages, setMessages] = useState([]);
  const [newMessage, setNewMessage] = useState('');
  const [isTyping, setIsTyping] = useState(false);
  const [onlineUsers, setOnlineUsers] = useState([]);
  const messagesEndRef = useRef(null);

  useEffect(() => {
    loadMessages();
    setupEventListeners();
    
    return () => {
      cleanup();
    };
  }, [roomId]);

  const loadMessages = async () => {
    try {
      const result = await client.service('messages').find({
        query: {
          roomId,
          $limit: 50,
          $sort: { createdAt: -1 }
        }
      });
      setMessages(result.data.reverse());
      scrollToBottom();
    } catch (error) {
      console.error('加载消息失败:', error);
    }
  };

  const setupEventListeners = () => {
    const messagesService = client.service('messages');
    
    messagesService.on('created', handleNewMessage);
    messagesService.on('patched', handleMessageUpdate);
    
    client.on('user-status-changed', handleUserStatusChange);
    client.on('user-typing', handleUserTyping);
  };

  const cleanup = () => {
    const messagesService = client.service('messages');
    messagesService.off('created', handleNewMessage);
    messagesService.off('patched', handleMessageUpdate);
    
    client.off('user-status-changed', handleUserStatusChange);
    client.off('user-typing', handleUserTyping);
  };

  const handleNewMessage = (message) => {
    if (message.roomId === roomId) {
      setMessages(prev => [...prev, message]);
      scrollToBottom();
      
      // 标记为已读
      if (message.senderId !== currentUser._id) {
        markAsRead([message._id]);
      }
    }
  };

  const handleMessageUpdate = (message) => {
    if (message.roomId === roomId) {
      setMessages(prev => 
        prev.map(msg => msg._id === message._id ? message : msg)
      );
    }
  };

  const handleUserStatusChange = (data) => {
    setOnlineUsers(prev => {
      const updated = prev.filter(user => user.userId !== data.userId);
      if (data.isOnline) {
        updated.push({
          userId: data.userId,
          activity: data.activity,
          lastSeen: data.lastSeen
        });
      }
      return updated;
    });
  };

  const handleUserTyping = (data) => {
    if (data.roomId === roomId && data.userId !== currentUser._id) {
      setIsTyping(data.isTyping);
      
      if (data.isTyping) {
        setTimeout(() => setIsTyping(false), 3000);
      }
    }
  };

  const sendMessage = async () => {
    if (!newMessage.trim()) return;

    try {
      await client.service('messages').create({
        roomId,
        content: newMessage,
        type: 'text'
      });
      
      setNewMessage('');
      stopTyping();
    } catch (error) {
      console.error('发送消息失败:', error);
    }
  };

  const markAsRead = async (messageIds) => {
    try {
      await client.service('messages').markAsRead(messageIds);
    } catch (error) {
      console.error('标记已读失败:', error);
    }
  };

  const startTyping = () => {
    client.emit('user-typing', {
      roomId,
      userId: currentUser._id,
      isTyping: true
    });
  };

  const stopTyping = () => {
    client.emit('user-typing', {
      roomId,
      userId: currentUser._id,
      isTyping: false
    });
  };

  const scrollToBottom = () => {
    messagesEndRef.current?.scrollIntoView({ behavior: 'smooth' });
  };

  const handleKeyPress = (e) => {
    if (e.key === 'Enter' && !e.shiftKey) {
      e.preventDefault();
      sendMessage();
    }
  };

  const handleInputChange = (e) => {
    setNewMessage(e.target.value);
    
    if (e.target.value.trim() && !isTyping) {
      startTyping();
    } else if (!e.target.value.trim() && isTyping) {
      stopTyping();
    }
  };

  return (
    <div className="chat-room">
      <div className="messages-container">
        {messages.map(message => (
          <MessageItem 
            key={message._id} 
            message={message}
            currentUser={currentUser}
            onReaction={(emoji) => addReaction(message._id, emoji)}
          />
        ))}
        {isTyping && (
          <div className="typing-indicator">
            有人正在输入...
          </div>
        )}
        <div ref={messagesEndRef} />
      </div>
      
      <div className="message-input">
        <textarea
          value={newMessage}
          onChange={handleInputChange}
          onKeyPress={handleKeyPress}
          placeholder="输入消息..."
          rows={1}
        />
        <button onClick={sendMessage} disabled={!newMessage.trim()}>
          发送
        </button>
      </div>
    </div>
  );
};

export default ChatRoom;

总结

通过这个聊天应用项目，我们实现了：

✅ 完整的用户系统：

• 用户注册登录
• 在线状态管理
• 联系人管理

✅ 强大的聊天功能：

• 私聊和群聊
• 多种消息类型
• 消息状态跟踪

✅ 实时通信：

• WebSocket 连接管理
• 实时消息同步
• 在线状态广播

✅ 高级特性：

• 消息搜索
• 反应表情
• 消息编辑和撤回

这个项目展示了 Feathers.js 在构建实时应用方面的强大能力，是学习和实践的绝佳案例。

---

相关文章推荐：

• Feathers.js 实时功能实战 - WebSocket 和事件系统
• Feathers.js 实战项目：博客系统 - 内容管理平台

有问题欢迎留言讨论，我会及时回复大家！