初始化

app/Amqp/Consumer/ChatMessageConsumer.php

@@ -4,6 +4,12 @@ declare(strict_types=1);
 
 namespace App\Amqp\Consumer;
 
+use App\Helper\PushMessageHelper;
+use App\Model\Chat\ChatRecord;
+use App\Model\Chat\ChatRecordsCode;
+use App\Model\Chat\ChatRecordsFile;
+use App\Service\SocketFDService;
+use App\Service\SocketRoomService;
 use Hyperf\Amqp\Result;
 use Hyperf\Amqp\Annotation\Consumer;
 use Hyperf\Amqp\Message\ConsumerMessage;
@@ -11,7 +17,6 @@ use Hyperf\Redis\Redis;
 use PhpAmqpLib\Message\AMQPMessage;
 use Hyperf\Amqp\Message\Type;
 use Hyperf\Amqp\Builder\QueueBuilder;
-use Hyperf\Utils\Coroutine;
 
 /**
  * @Consumer(name="聊天消息消费者",enable=true)
@@ -40,10 +45,23 @@ class ChatMessageConsumer extends ConsumerMessage
     public $routingKey = 'consumer:im:message';
 
     /**
-     * ImMessageConsumer constructor.
+     * @var SocketFDService
      */
-    public function __construct()
+    private $socketFDService;
+
+    /**
+     * @var SocketRoomService
+     */
+    private $socketRoomService;
+
+    /**
+     * ChatMessageConsumer constructor.
+     * @param SocketFDService $socketFDService
+     */
+    public function __construct(SocketFDService $socketFDService, SocketRoomService $socketRoomService)
     {
+        $this->socketFDService = $socketFDService;
+        $this->socketRoomService = $socketRoomService;
         $this->setQueue('queue:im-message:' . SERVER_RUN_ID);
     }
 
@@ -68,8 +86,6 @@ class ChatMessageConsumer extends ConsumerMessage
      */
     public function consumeMessage($data, AMQPMessage $message): string
     {
-
-        go(function() use ($data) {
         $redis = container()->get(Redis::class);
 
         //[加锁]防止消息重复消费
@@ -78,15 +94,81 @@ class ChatMessageConsumer extends ConsumerMessage
             return Result::ACK;
         }
 
+        $source = $data['source'];
+
+        $fids = $this->socketFDService->findUserFds($data['sender']);
+        if ($source == 1) {// 私聊
+            $fids = array_merge($fids, $this->socketFDService->findUserFds($data['receive']));
+        } else if ($source == 2) {//群聊
+            $userIds = $this->socketRoomService->getRoomMembers(strval($data['receive']));
+            foreach ($userIds as $uid) {
+                $fids = array_merge($fids, $this->socketFDService->findUserFds(intval($uid)));
+            }
+        }
+
+        // 去重
+        $fids = array_unique($fids);
+        if (empty($fids)) {
+            return Result::ACK;
+        }
+
+        $result = ChatRecord::leftJoin('users', 'users.id', '=', 'chat_records.user_id')
+            ->where('chat_records.id', $data['record_id'])
+            ->first([
+                'chat_records.id',
+                'chat_records.source',
+                'chat_records.msg_type',
+                'chat_records.user_id',
+                'chat_records.receive_id',
+                'chat_records.content',
+                'chat_records.is_revoke',
+                'chat_records.created_at',
+
+                'users.nickname',
+                'users.avatar as avatar',
+            ]);
+
+        $file = [];
+        $code_block = [];
+        if ($result->msg_type == 2) {
+            $file = ChatRecordsFile::where('record_id', $result->id)->first(['id', 'record_id', 'user_id', 'file_source', 'file_type', 'save_type', 'original_name', 'file_suffix', 'file_size', 'save_dir']);
+            $file = $file ? $file->toArray() : [];
+            if ($file) {
+                $file['file_url'] = get_media_url($file['save_dir']);
+            }
+        } else if ($result->msg_type == 5) {
+            $code_block = ChatRecordsCode::where('record_id', $result->id)->first(['record_id', 'code_lang', 'code']);
+            $code_block = $code_block ? $code_block->toArray() : [];
+        }
+
+        $msg = [
+            'send_user' => $data['sender'],
+            'receive_user' => $data['receive'],
+            'source_type' => $data['source'],
+            'data' => PushMessageHelper::formatTalkMsg([
+                'id' => $result->id,
+                'msg_type' => $result->msg_type,
+                'source' => $result->source,
+                'avatar' => $result->avatar,
+                'nickname' => $result->nickname,
+                "user_id" => $result->user_id,
+                "receive_id" => $result->receive_id,
+                "created_at" => $result->created_at,
+                "content" => $result->content,
+                "file" => $file,
+                "code_block" => $code_block
+            ])
+        ];
+
         $server = server();
-            foreach ($server->connections as $fd) {
+        foreach ($fids as $fd) {
-                if ($server->exist($fd) && $server->isEstablished($fd)) {
+            $fd = intval($fd);
-                    $server->push($fd, "Recv: 我是后台进程 [{$data['message']}]");
+            if ($server->exist($fd)) {
+                $server->push($fd, json_encode(['chat_message', $msg]));
             }
         }
-        });
-
 
+        unset($fids, $result, $msg);
         return Result::ACK;
     }
 }

app/Amqp/Producer/ChatMessageProducer.php

@@ -14,15 +14,14 @@ class ChatMessageProducer extends ProducerMessage
 
     public $type = Type::FANOUT;
 
-    public function __construct($data)
+    public function __construct($sender, $receive, $source, $record_id)
     {
         $message = [
             'uuid' => $this->uuid(),
-            'method' => '', //
+            'sender' => intval($sender),  //发送者ID
-            'sender' => '',  //发送者ID
+            'receive' => intval($receive),  //接收者ID
-            'receive' => '',  //接收者ID
+            'source' => intval($source), //接收者类型 1:好友;2:群组
-            'receiveType' => '', //接收者类型 1:好友;2:群组
+            'record_id' => intval($record_id)
-            'message' => $data
         ];
 
         $this->payload = $message;

app/Controller/Api/V1/ArticleController.php

@@ -1,4 +1,5 @@
 <?php
+declare(strict_types=1);
 
 namespace App\Controller\Api\V1;
 
@@ -82,7 +83,7 @@ class ArticleController extends CController
         }
 
         return $this->response->success(
-            $this->articleService->getUserArticleList($user_id, $page, 10000, $params)
+            $this->articleService->getUserArticleList($user_id, intval($page), 10000, $params)
         );
     }
 

app/Controller/Api/V1/DownloadController.php

@@ -0,0 +1,63 @@
+<?php
+
+
+namespace App\Controller\Api\V1;
+
+use App\Model\Chat\ChatRecord;
+use App\Model\Chat\ChatRecordsFile;
+use App\Model\Group\UsersGroup;
+use App\Service\UploadService;
+use Hyperf\Di\Annotation\Inject;
+use Hyperf\HttpServer\Annotation\Controller;
+use Hyperf\HttpServer\Annotation\RequestMapping;
+use Hyperf\HttpServer\Annotation\Middleware;
+use Hyperf\HttpServer\Contract\ResponseInterface;
+use App\Middleware\JWTAuthMiddleware;
+
+/**
+ * Class DownloadController
+ *
+ * @Controller(path="/api/v1/download")
+ *
+ * @package App\Controller\Api\V1
+ */
+class DownloadController extends CController
+{
+    /**
+     * 下载用户聊天文件
+     *
+     * @RequestMapping(path="user-chat-file", methods="get")
+     *
+     * @return mixed
+     */
+    public function userChatFile(ResponseInterface $response, UploadService $uploadService)
+    {
+        $crId = $this->request->input('cr_id', 0);
+        $uid = 2054;
+
+        $recordsInfo = ChatRecord::select(['msg_type', 'source', 'user_id', 'receive_id'])->where('id', $crId)->first();
+        if (!$recordsInfo) {
+            return $this->response->fail('文件不存在...');
+        }
+
+        //判断消息是否是当前用户发送(如果是则跳过权限验证)
+        if ($recordsInfo->user_id != $uid) {
+            if ($recordsInfo->source == 1) {
+                if ($recordsInfo->receive_id != $uid) {
+                    return $this->response->fail('非法请求...');
+                }
+            } else {
+                if (!UsersGroup::isMember($recordsInfo->receive_id, $uid)) {
+                    return $this->response->fail('非法请求...');
+                }
+            }
+        }
+
+        $fileInfo = ChatRecordsFile::select(['save_dir', 'original_name'])->where('record_id', $crId)->first();
+        if (!$fileInfo) {
+            return $this->response->fail('文件不存在或没有下载权限...');
+        }
+
+        return $response->download($uploadService->driver($fileInfo->save_dir), $fileInfo->original_name);
+    }
+}

app/Controller/Api/V1/TalkController.php

@@ -4,16 +4,23 @@ namespace App\Controller\Api\V1;
 
 use App\Cache\LastMsgCache;
 use App\Cache\UnreadTalkCache;
+use App\Model\EmoticonDetail;
+use App\Model\FileSplitUpload;
 use App\Model\User;
 use App\Model\UsersChatList;
 use App\Model\UsersFriend;
 use App\Model\Group\UsersGroup;
 use App\Service\TalkService;
+use App\Service\UploadService;
+use Hyperf\Amqp\Producer;
 use Hyperf\Di\Annotation\Inject;
 use Hyperf\HttpServer\Annotation\Controller;
 use Hyperf\HttpServer\Annotation\RequestMapping;
 use Hyperf\HttpServer\Annotation\Middleware;
 use Phper666\JWTAuth\Middleware\JWTAuthMiddleware;
+use Hyperf\Utils\Str;
+use Psr\Http\Message\ResponseInterface;
+use App\Amqp\Producer\ChatMessageProducer;
 
 /**
  * Class TalkController
@@ -37,6 +44,12 @@ class TalkController extends CController
      */
     public $unreadTalkCache;
 
+    /**
+     * @Inject
+     * @var Producer
+     */
+    private $producer;
+
     /**
      * 获取用户对话列表
      *
@@ -437,10 +450,64 @@ class TalkController extends CController
      * 上传聊天对话图片（待优化）
      *
      * @RequestMapping(path="send-image", methods="post")
+     *
+     * @param UploadService $uploadService
+     * @return ResponseInterface
      */
-    public function sendImage()
+    public function sendImage(UploadService $uploadService)
     {
+        $params = $this->request->inputs(['source', 'receive_id']);
+        $this->validate($params, [
+            //消息来源（1：好友消息 2：群聊消息）
+            'source' => 'required|in:1,2',
+            'receive_id' => 'required|integer|min:1'
+        ]);
 
+        $file = $this->request->file('img');
+        if (!$file->isValid()) {
+            return $this->response->fail();
+        }
+
+        $ext = $file->getExtension();
+        if (!in_array($ext, ['jpg', 'png', 'jpeg', 'gif', 'webp'])) {
+            return $this->response->fail('图片格式错误，目前仅支持jpg、png、jpeg、gif和webp');
+        }
+
+        //获取图片信息
+        $imgInfo = getimagesize($file->getRealPath());
+
+        $path = $uploadService->media($file, 'media/images/talks', create_image_name($ext, $imgInfo[0], $imgInfo[1]));
+        if (!$path) {
+            return $this->response->fail();
+        }
+
+        $user_id = $this->uid();
+
+        // 创建图片消息记录
+        $record_id = $this->talkService->createImgMessage([
+            'source' => $params['source'],
+            'msg_type' => 2,
+            'user_id' => $user_id,
+            'receive_id' => $params['receive_id'],
+        ], [
+            'user_id' => $user_id,
+            'file_type' => 1,
+            'file_suffix' => $ext,
+            'file_size' => $file->getSize(),
+            'save_dir' => $path,
+            'original_name' => $file->getClientFilename(),
+        ]);
+
+        if (!$record_id) {
+            return $this->response->fail('图片上传失败');
+        }
+
+        // ...消息推送队列
+        $this->producer->produce(
+            new ChatMessageProducer($user_id, $params['receive_id'], $params['source'], $record_id)
+        );
+
+        return $this->response->success();
     }
 
     /**
@@ -450,7 +517,37 @@ class TalkController extends CController
      */
     public function sendCodeBlock()
     {
+        $params = $this->request->inputs(['source', 'receive_id', 'lang', 'code']);
+        $this->validate($params, [
+            //消息来源（1：好友消息 2：群聊消息）
+            'source' => 'required|in:1,2',
+            'receive_id' => 'required|integer|min:1',
+            'lang' => 'required',
+            'code' => 'required'
+        ]);
 
+        $user_id = $this->uid();
+        $record_id = $this->talkService->createCodeMessage([
+            'source' => $params['source'],
+            'msg_type' => 5,
+            'user_id' => $user_id,
+            'receive_id' => $params['receive_id'],
+        ], [
+            'user_id' => $user_id,
+            'code_lang' => $params['lang'],
+            'code' => $params['code']
+        ]);
+
+        if (!$record_id) {
+            return $this->response->fail('消息发送失败');
+        }
+
+        // ...消息推送队列
+        $this->producer->produce(
+            new ChatMessageProducer($user_id, $params['receive_id'], $params['source'], $record_id)
+        );
+
+        return $this->response->success();
     }
 
     /**
@@ -458,9 +555,57 @@ class TalkController extends CController
      *
      * @RequestMapping(path="send-file", methods="post")
      */
-    public function sendFile()
+    public function sendFile(UploadService $uploadService)
     {
+        $params = $this->request->inputs(['hash_name', 'receive_id', 'source']);
+        $this->validate($params, [
+            //消息来源（1：好友消息 2：群聊消息）
+            'source' => 'required|in:1,2',
+            'receive_id' => 'required|integer|min:1',
+            'hash_name' => 'required',
+        ]);
 
+        $user_id = $this->uid();
+
+        $file = FileSplitUpload::where('user_id', $user_id)->where('hash_name', $params['hash_name'])->where('file_type', 1)->first();
+        if (!$file || empty($file->save_dir)) {
+            return $this->response->fail('文件不存在...');
+        }
+
+        $file_hash_name = uniqid() . Str::random(10) . '.' . $file->file_ext;
+        $save_dir = "files/talks/" . date('Ymd') . '/' . $file_hash_name;
+
+        $uploadService->makeDirectory($uploadService->driver("files/talks/" . date('Ymd')));
+
+        // Copy Files
+        @copy($uploadService->driver($file->save_dir), $uploadService->driver($save_dir));
+
+        $record_id = $this->talkService->createFileMessage([
+            'source' => $params['source'],
+            'msg_type' => 2,
+            'user_id' => $user_id,
+            'receive_id' => $params['receive_id']
+        ], [
+            'user_id' => $user_id,
+            'file_source' => 1,
+            'file_type' => 4,
+            'original_name' => $file->original_name,
+            'file_suffix' => $file->file_ext,
+            'file_size' => $file->file_size,
+            'save_dir' => $save_dir,
+        ]);
+
+        if (!$record_id) {
+            return $this->response->fail('表情发送失败');
+        }
+
+        // ...消息推送队列
+        $this->producer->produce(
+            new ChatMessageProducer($user_id, $params['receive_id'], $params['source'], $record_id)
+        );
+
+
+        return $this->response->success();
     }
 
     /**
@@ -470,6 +615,48 @@ class TalkController extends CController
      */
     public function sendEmoticon()
     {
+        $params = $this->request->inputs(['source', 'receive_id', 'emoticon_id']);
+        $this->validate($params, [
+            //消息来源（1：好友消息 2：群聊消息）
+            'source' => 'required|in:1,2',
+            'receive_id' => 'required|integer|min:1',
+            'emoticon_id' => 'required|integer|min:1',
+        ]);
 
+        $user_id = $this->uid();
+        $emoticon = EmoticonDetail::where('id', $params['emoticon_id'])->where('user_id', $user_id)->first([
+            'url',
+            'file_suffix',
+            'file_size'
+        ]);
+
+        if (!$emoticon) {
+            return $this->response->fail('表情不存在...');
+        }
+
+        $record_id = $this->talkService->createEmoticonMessage([
+            'source' => $params['source'],
+            'msg_type' => 2,
+            'user_id' => $user_id,
+            'receive_id' => $params['receive_id'],
+        ], [
+            'user_id' => $user_id,
+            'file_type' => 1,
+            'file_suffix' => $emoticon->file_suffix,
+            'file_size' => $emoticon->file_size,
+            'save_dir' => $emoticon->url,
+            'original_name' => '表情',
+        ]);
+
+        if (!$record_id) {
+            return $this->response->fail('表情发送失败');
+        }
+
+        // ...消息推送队列
+        $this->producer->produce(
+            new ChatMessageProducer($user_id, $params['receive_id'], $params['source'], $record_id)
+        );
+
+        return $this->response->success();
     }
 }

app/Controller/Api/V1/UploadController.php

@@ -0,0 +1,100 @@
+<?php
+declare(strict_types=1);
+
+namespace App\Controller\Api\V1;
+
+use App\Service\SplitUploadService;
+use App\Service\UploadService;
+use Hyperf\Di\Annotation\Inject;
+use Hyperf\HttpServer\Annotation\Controller;
+use Hyperf\HttpServer\Annotation\RequestMapping;
+use Hyperf\HttpServer\Annotation\Middleware;
+use Phper666\JWTAuth\Middleware\JWTAuthMiddleware;
+
+/**
+ * 上传控制器
+ *
+ * Class UploadController
+ *
+ * @Controller(path="/api/v1/upload")
+ * @Middleware(JWTAuthMiddleware::class)
+ *
+ * @package App\Controller\Api\V1
+ */
+class UploadController extends CController
+{
+    /**
+     * @inject
+     * @var UploadService
+     */
+    private $uploadService;
+
+    /**
+     * @inject
+     * @var SplitUploadService
+     */
+    private $splitUploadService;
+
+    /**
+     * 获取拆分文件信息
+     *
+     * @RequestMapping(path="get-file-split-info", methods="get")
+     *
+     */
+    public function getFileSplitInfo()
+    {
+        $params = $this->request->inputs(['file_name', 'file_size']);
+        $this->validate($params, [
+            'file_name' => "required",
+            'file_size' => 'required|integer'
+        ]);
+
+        $data = $this->splitUploadService->create($this->uid(), $params['file_name'], $params['file_size']);
+
+        return $data ? $this->response->success($data) : $this->response->fail('获取文件拆分信息失败...');
+    }
+
+    /**
+     *
+     * @RequestMapping(path="file-subarea-upload", methods="post")
+     *
+     * @return \Psr\Http\Message\ResponseInterface
+     */
+    public function fileSubareaUpload()
+    {
+        $file = $this->request->file('file');
+        $params = $this->request->inputs(['name', 'hash', 'ext', 'size', 'split_index', 'split_num']);
+        $this->validate($params, [
+            'name' => "required",
+            'hash' => 'required',
+            'ext' => 'required',
+            'size' => 'required',
+            'split_index' => 'required',
+            'split_num' => 'required'
+        ]);
+
+        if (!$file->isValid()) {
+            return $this->response->fail();
+        }
+
+        $user_id = $this->uid();
+        $uploadRes = $this->splitUploadService->upload($user_id, $file, $params['hash'], intval($params['split_index']), intval($params['size']));
+        if (!$uploadRes) {
+            return $this->response->fail('上传文件失败...');
+        }
+
+        if (($params['split_index'] + 1) == $params['split_num']) {
+            $fileInfo = $this->splitUploadService->merge($user_id, $params['hash']);
+            if (!$fileInfo) {
+                return $this->response->fail('上传文件失败...');
+            }
+
+            return $this->response->success([
+                'is_file_merge' => true,
+                'hash' => $params['hash']
+            ]);
+        }
+
+        return $this->response->success(['is_file_merge' => false]);
+    }
+}

app/Controller/WebSocketController.php

@@ -3,11 +3,15 @@ declare(strict_types=1);
 
 namespace App\Controller;
 
+use App\Cache\LastMsgCache;
+use App\Cache\UnreadTalkCache;
+use App\Model\Chat\ChatRecord;
+use App\Model\Group\UsersGroupMember;
+use App\Service\SocketRoomService;
 use Hyperf\Di\Annotation\Inject;
 use Hyperf\Contract\OnCloseInterface;
 use Hyperf\Contract\OnMessageInterface;
 use Hyperf\Contract\OnOpenInterface;
-use Hyperf\Utils\Codec\Json;
 use Phper666\JWTAuth\JWT;
 use Swoole\Http\Request;
 use Swoole\Websocket\Frame;
@@ -44,6 +48,12 @@ class WebSocketController implements OnMessageInterface, OnOpenInterface, OnClos
      */
     private $socketFDService;
 
+    /**
+     * @inject
+     * @var SocketRoomService
+     */
+    private $socketRoomService;
+
     /**
      * 连接创建成功回调事件
      *
@@ -54,14 +64,16 @@ class WebSocketController implements OnMessageInterface, OnOpenInterface, OnClos
     {
         $token = $request->get['token'] ?? '';
         $userInfo = $this->jwt->getParserData($token);
-        stdout_log()->notice("用户连接信息 : user_id:{$userInfo['user_id']} | fd:{$request->fd} | data:" . Json::encode($userInfo));
+        stdout_log()->notice("用户连接信息 : user_id:{$userInfo['user_id']} | fd:{$request->fd} 时间：" . date('Y-m-d H:i:s'));
-        stdout_log()->notice('连接时间：' . date('Y-m-d H:i:s'));
 
         // 绑定fd与用户关系
         $this->socketFDService->bindRelation($request->fd, $userInfo['user_id']);
 
-        $ip = config('ip_address');
+        // 加入群聊
-        $server->push($request->fd, "成功连接[{$ip}],IM 服务器");
+        $groupIds = UsersGroupMember::getUserGroupIds($userInfo['user_id']);
+        foreach ($groupIds as $group_id) {
+            $this->socketRoomService->addRoomMember($userInfo['user_id'], $group_id);
+        }
     }
 
     /**
@@ -75,7 +87,59 @@ class WebSocketController implements OnMessageInterface, OnOpenInterface, OnClos
         // 判断是否为心跳检测
         if ($frame->data == 'PING') return;
 
-        $this->producer->produce(new ChatMessageProducer("我是来自 xxxx 服务器的消息]，{$frame->data}"));
+        // 当前用户ID
+        $user_id = $this->socketFDService->findFdUserId($frame->fd);
+
+        [$event, $data] = array_values(json_decode($frame->data, true));
+        if ($user_id != $data['send_user']) {
+            return;
+        }
+
+        //验证消息类型 私聊|群聊
+        if (!in_array($data['source_type'], [1, 2])) {
+            return;
+        }
+
+        //验证发送消息用户与接受消息用户之间是否存在好友或群聊关系(后期走缓存)
+//        if ($data['source_type'] == 1) {//私信
+//            //判断发送者和接受者是否是好友关系
+//            if (!UsersFriend::isFriend(intval($data['send_user']), intval($data['receive_user']))) {
+//                return;
+//            }
+//        } else if ($data['source_type'] == 2) {//群聊
+//            //判断是否属于群成员
+//            if (!UsersGroup::isMember(intval($data['receive_user']), intval($data['send_user']))) {
+//                return;
+//            }
+//        }
+
+        $result = ChatRecord::create([
+            'source' => $data['source_type'],
+            'msg_type' => 1,
+            'user_id' => $data['send_user'],
+            'receive_id' => $data['receive_user'],
+            'content' => htmlspecialchars($data['text_message']),
+            'created_at' => date('Y-m-d H:i:s'),
+        ]);
+
+        if (!$result) return;
+
+        // 判断是否私聊
+        if ($data['source_type'] == 1) {
+            $msg_text = mb_substr($result->content, 0, 30);
+            // 缓存最后一条消息
+            LastMsgCache::set([
+                'text' => $msg_text,
+                'created_at' => $result->created_at
+            ], intval($data['receive_user']), intval($data['send_user']));
+
+            // 设置好友消息未读数
+            make(UnreadTalkCache::class)->setInc(intval($result->receive_id), strval($result->user_id));
+        }
+
+        $this->producer->produce(
+            new ChatMessageProducer($data['send_user'], $data['receive_user'], $data['source_type'], $result->id)
+        );
     }
 
     /**

app/Helper/PushMessageHelper.php

@@ -0,0 +1,68 @@
+<?php
+
+namespace App\Helper;
+
+
+use App\Model\User;
+
+/**
+ * Socket 资源处理
+ * Class PushMessageHelper
+ * @package App\Helpers
+ */
+class PushMessageHelper
+{
+    // 消息事件类型
+    const events = [
+        'chat_message',//用户聊天消息
+        'friend_apply',//好友添加申请消息
+        'join_group',    //入群消息
+        'login_notify',//好友登录消息通知
+        'input_tip',//好友登录消息通知
+        'revoke_records',//好友撤回消息通知
+    ];
+
+    /**
+     * 格式化对话的消息体
+     *
+     * @param array $data 对话的消息
+     * @return array
+     */
+    public static function formatTalkMsg(array $data)
+    {
+        // 缓存优化
+        if (!isset($data['nickname']) || !isset($data['avatar']) || empty($data['nickname']) || empty($data['avatar'])) {
+            if (isset($data['user_id']) && !empty($data['user_id'])) {
+                $info = User::where('id', $data['user_id'])->first(['nickname', 'avatar']);
+                if ($info) {
+                    $data['nickname'] = $info->nickname;
+                    $data['avatar'] = $info->avatar;
+                }
+            }
+        }
+
+        $arr = [
+            "id" => 0,
+            "source" => 1,
+            "msg_type" => 1,
+            "user_id" => 0,
+            "receive_id" => 0,
+            "content" => '',
+            "is_revoke" => 0,
+
+            // 发送消息人的信息
+            "nickname" => "",
+            "avatar" => "",
+
+            // 不同的消息类型
+            "file" => [],
+            "code_block" => [],
+            "forward" => [],
+            "invite" => [],
+
+            "created_at" => "",
+        ];
+
+        return array_merge($arr, array_intersect_key($data, $arr));
+    }
+}

app/Model/Chat/ChatRecord.php

@@ -55,7 +55,6 @@ class ChatRecord extends BaseModel
         'msg_type' => 'integer',
         'user_id' => 'integer',
         'receive_id' => 'integer',
-        'is_revoke' => 'integer',
+        'is_revoke' => 'integer'
-        'created_at' => 'datetime'
     ];
 }

app/Model/Chat/ChatRecordsCode.php

@@ -32,7 +32,13 @@ class ChatRecordsCode extends BaseModel
      *
      * @var array
      */
-    protected $fillable = [];
+    protected $fillable = [
+        'record_id',
+        'user_id',
+        'code_lang',
+        'code',
+        'created_at'
+    ];
 
     /**
      * The attributes that should be cast to native types.

app/Model/Chat/ChatRecordsFile.php

@@ -38,12 +38,34 @@ class ChatRecordsFile extends BaseModel
      *
      * @var array
      */
-    protected $fillable = [];
+    protected $fillable = [
+        'record_id',
+        'user_id',
+        'file_source',
+        'file_type',
+        'save_type',
+        'original_name',
+        'file_suffix',
+        'file_size',
+        'save_dir',
+        'is_delete',
+        'created_at'
+    ];
     
     /**
      * The attributes that should be cast to native types.
      *
      * @var array
      */
-    protected $casts = ['id' => 'integer', 'record_id' => 'integer', 'user_id' => 'integer', 'file_source' => 'integer', 'file_type' => 'integer', 'save_type' => 'integer', 'file_size' => 'integer', 'is_delete' => 'integer', 'created_at' => 'datetime'];
+    protected $casts = [
+        'id' => 'integer',
+        'record_id' => 'integer',
+        'user_id' => 'integer',
+        'file_source' => 'integer',
+        'file_type' => 'integer',
+        'save_type' => 'integer',
+        'file_size' => 'integer',
+        'is_delete' => 'integer',
+        'created_at' => 'datetime'
+    ];
 }

app/Model/FileSplitUpload.php

@@ -37,9 +37,20 @@ class FileSplitUpload extends BaseModel
      * @var array
      */
     protected $fillable = [
-
+        'file_type',
+        'user_id',
+        'hash_name',
+        'original_name',
+        'split_index',
+        'split_num',
+        'save_dir',
+        'file_ext',
+        'file_size',
+        'is_delete',
+        'upload_at'
     ];
 
+
     /**
      * The attributes that should be cast to native types.
      *

app/Model/Group/UsersGroupMember.php

@@ -71,4 +71,15 @@ class UsersGroupMember extends BaseModel
     {
         return self::where('group_id', $group_id)->where('user_id', $user_id)->value('visit_card');
     }
+
+    /**
+     * 获取用户的所有群ID
+     *
+     * @param int $user_id
+     * @return array
+     */
+    public static function getUserGroupIds(int $user_id)
+    {
+        return self::where('user_id', $user_id)->where('status', 0)->pluck('group_id')->toArray();
+    }
 }

app/Model/UsersChatList.php

@@ -34,7 +34,17 @@ class UsersChatList extends BaseModel
      *
      * @var array
      */
-    protected $fillable = [];
+    protected $fillable = [
+        'type',
+        'uid',
+        'friend_id',
+        'group_id',
+        'status',
+        'is_top',
+        'not_disturb',
+        'created_at',
+        'updated_at'
+    ];
 
     /**
      * The attributes that should be cast to native types.

app/Service/EmoticonService.php

@@ -80,7 +80,7 @@ class EmoticonService extends BaseService
     public function getInstallIds(int $user_id)
     {
         $result = UsersEmoticon::where('user_id', $user_id)->value('emoticon_ids');
-        return $result ? explode(',', $result) : [];
+        return $result ? array_filter($result) : [];
     }
 
     /**

app/Service/SocketRoomService.php

@@ -38,7 +38,7 @@ class SocketRoomService
      */
     public function addRoomMember(int $usr_id, $room)
     {
-        return redis()->sAdd($this->getRoomName($room), $room);
+        return redis()->sAdd($this->getRoomName($room), $usr_id);
     }
 
     /**

app/Service/SplitUploadService.php

@@ -0,0 +1,133 @@
+<?php
+
+
+namespace App\Service;
+
+use App\Model\FileSplitUpload;
+use Hyperf\Utils\Str;
+use Hyperf\HttpMessage\Upload\UploadedFile;
+
+/**
+ * 文件拆分上传服务
+ *
+ * Class SplitUploadService
+ * @package App\Service
+ */
+class SplitUploadService
+{
+    const SPLIT_SIZE = 2 * 1024 * 1024;
+
+    /**
+     * 创建文件拆分相关信息
+     *
+     * @param int $user_id 用户ID
+     * @param string $fileName 上传的文件名
+     * @param string $fileSize 上传文件大小
+     */
+    public function create(int $user_id, string $fileName, string $fileSize)
+    {
+        $hash_name = implode('-', [uniqid(), rand(10000000, 99999999), Str::random(6)]);
+        $split_num = intval(ceil($fileSize / self::SPLIT_SIZE));
+
+        $data = [];
+        $data['file_type'] = 1;
+        $data['user_id'] = $user_id;
+        $data['original_name'] = $fileName;
+        $data['hash_name'] = $hash_name;
+        $data['file_ext'] = pathinfo($fileName, PATHINFO_EXTENSION);
+        $data['file_size'] = $fileSize;
+        $data['upload_at'] = time();
+
+        //文件拆分数量
+        $data['split_num'] = $split_num;
+        $data['split_index'] = $split_num;
+
+        return FileSplitUpload::create($data) ? array_merge($data, ['split_size' => self::SPLIT_SIZE]) : false;
+    }
+
+    /**
+     * @param int $user_id
+     * @param UploadedFile $file 文件信息
+     * @param string $hashName 上传临时问价hash名
+     * @param int $split_index 当前拆分文件索引
+     * @param int $fileSize 文件大小
+     */
+    public function upload(int $user_id, UploadedFile $file, string $hashName, int $split_index, int $fileSize)
+    {
+        $fileInfo = FileSplitUpload::select(['id', 'original_name', 'split_num', 'file_ext'])
+            ->where([['user_id', '=', $user_id], ['hash_name', '=', $hashName], ['file_type', '=', 1]])
+            ->first();
+
+        if (!$fileInfo) {
+            return false;
+        }
+
+        // 保存文件名及保存文件相对目录
+        $fileName = "{$hashName}_{$split_index}_{$fileInfo->file_ext}.tmp";
+        $uploadService = make(UploadService::class);
+
+        $uploadService->makeDirectory($uploadService->driver("tmp/{$hashName}"));
+        $file->moveTo(sprintf('%s/%s', $uploadService->driver("tmp/{$hashName}"), $fileName));
+
+        if (!$file->isMoved()) {
+            return false;
+        }
+
+        $info = FileSplitUpload::where('user_id', $user_id)->where('hash_name', $hashName)->where('split_index', $split_index)->first();
+        if (!$info) {
+            return FileSplitUpload::create([
+                'user_id' => $user_id,
+                'file_type' => 2,
+                'hash_name' => $hashName,
+                'original_name' => $fileInfo->original_name,
+                'split_index' => $split_index,
+                'split_num' => $fileInfo->split_num,
+                'save_dir' => sprintf('%s/%s', "tmp/{$hashName}", $fileName),
+                'file_ext' => $fileInfo->file_ext,
+                'file_size' => $fileSize,
+                'upload_at' => time(),
+            ]) ? true : false;
+        }
+
+        return true;
+    }
+
+    /**
+     * 文件合并
+     *
+     * @param string $hash_name 上传临时问价hash名
+     */
+    public function merge(int $user_id, string $hash_name)
+    {
+        $fileInfo = FileSplitUpload::select(['id', 'original_name', 'split_num', 'file_ext', 'file_size'])->where('user_id', $user_id)->where('hash_name', $hash_name)->where('file_type', 1)->first();
+        if (!$fileInfo) {
+            return false;
+        }
+
+        $files = FileSplitUpload::where('user_id', $user_id)->where('hash_name', $hash_name)->where('file_type', 2)->orderBy('split_index', 'asc')->get(['split_index', 'save_dir'])->toArray();
+        if (!$files) {
+            return false;
+        }
+
+        if (count($files) != $fileInfo->split_num) {
+            return false;
+        }
+
+        $dir = config('filesystems.disks.uploads.root');
+        $fileMerge = "tmp/{$hash_name}/{$fileInfo->original_name}.tmp";
+        $uploadService = make(UploadService::class);
+
+        $merge_save_parh = $uploadService->driver($fileMerge);
+        foreach ($files as $file) {
+            file_put_contents($merge_save_parh, file_get_contents($uploadService->driver($file['save_dir'])), FILE_APPEND);
+        }
+
+        FileSplitUpload::select(['id', 'original_name', 'split_num', 'file_ext', 'file_size'])->where('user_id', $user_id)->where('hash_name', $hash_name)->where('file_type', 1)->update(['save_dir' => $fileMerge]);
+        return [
+            'path' => $fileMerge,
+            'tmp_file_name' => "{$fileInfo->original_name}.tmp",
+            'original_name' => $fileInfo->original_name,
+            'file_size' => $fileInfo->file_size
+        ];
+    }
+}

app/Service/TalkService.php

@@ -672,4 +672,129 @@ class TalkService extends BaseService
         $rows = $rowsSqlObj->orderBy('chat_records.id', 'desc')->forPage($page, $page_size)->get()->toArray();
         return $this->getPagingRows($this->handleChatRecords($rows), $count, $page, $page_size);
     }
+
+    /**
+     * 创建图片消息
+     *
+     * @param $message
+     * @param $fileInfo
+     */
+    public function createImgMessage($message, $fileInfo)
+    {
+        Db::beginTransaction();
+        try {
+            $message['created_at'] = date('Y-m-d H:i:s');
+            $insert = ChatRecord::create($message);
+
+            if (!$insert) {
+                throw new \Exception('插入聊天记录失败...');
+            }
+
+            $fileInfo['record_id'] = $insert->id;
+            $fileInfo['created_at'] = date('Y-m-d H:i:s');
+            if (!ChatRecordsFile::create($fileInfo)) {
+                throw new \Exception('插入聊天记录(文件消息)失败...');
+            }
+
+            Db::commit();
+        } catch (\Exception $e) {
+            Db::rollBack();
+            return false;
+        }
+
+        return $insert->id;
+    }
+
+    /**
+     * 创建代码块消息
+     *
+     * @param array $message
+     * @param array $codeBlock
+     */
+    public function createCodeMessage(array $message, array $codeBlock)
+    {
+        Db::beginTransaction();
+        try {
+            $message['created_at'] = date('Y-m-d H:i:s');
+            $insert = ChatRecord::create($message);
+            if (!$insert) {
+                throw new \Exception('插入聊天记录失败...');
+            }
+
+            $codeBlock['record_id'] = $insert->id;
+            $codeBlock['created_at'] = date('Y-m-d H:i:s');
+            if (!ChatRecordsCode::create($codeBlock)) {
+                throw new \Exception('插入聊天记录(代码消息)失败...');
+            }
+
+            Db::commit();
+        } catch (\Exception $e) {
+            Db::rollBack();
+            return false;
+        }
+
+        return $insert->id;
+    }
+
+    /**
+     * 创建代码块消息
+     *
+     * @param array $message
+     * @param array $emoticon
+     */
+    public function createEmoticonMessage(array $message, array $emoticon)
+    {
+        Db::beginTransaction();
+        try {
+            $message['created_at'] = date('Y-m-d H:i:s');
+            $insert = ChatRecord::create($message);
+            if (!$insert) {
+                throw new \Exception('插入聊天记录失败...');
+            }
+
+            $emoticon['record_id'] = $insert->id;
+            $emoticon['created_at'] = date('Y-m-d H:i:s');
+            if (!ChatRecordsFile::create($emoticon)) {
+                throw new \Exception('插入聊天记录(代码消息)失败...');
+            }
+
+            Db::commit();
+        } catch (\Exception $e) {
+            Db::rollBack();
+            return false;
+        }
+
+        return $insert->id;
+    }
+
+    /**
+     * 创建代码块消息
+     *
+     * @param array $message
+     * @param array $emoticon
+     */
+    public function createFileMessage(array $message, array $emoticon)
+    {
+        Db::beginTransaction();
+        try {
+            $message['created_at'] = date('Y-m-d H:i:s');
+            $insert = ChatRecord::create($message);
+            if (!$insert) {
+                throw new \Exception('插入聊天记录失败...');
+            }
+
+            $emoticon['record_id'] = $insert->id;
+            $emoticon['created_at'] = date('Y-m-d H:i:s');
+            if (!ChatRecordsFile::create($emoticon)) {
+                throw new \Exception('插入聊天记录(代码消息)失败...');
+            }
+
+            Db::commit();
+        } catch (\Exception $e) {
+            Db::rollBack();
+            return false;
+        }
+
+        return $insert->id;
+    }
 }

app/Service/UploadService.php

@@ -0,0 +1,47 @@
+<?php
+
+
+namespace App\Service;
+
+use Hyperf\HttpMessage\Upload\UploadedFile;
+use League\Flysystem\Filesystem;
+
+/**
+ * 文件上传服务
+ *
+ * Class UploadService
+ * @package App\Service
+ */
+class UploadService extends BaseService
+{
+    public function driver($dir)
+    {
+        return sprintf('%s/%s', rtrim(config('upload_dir'), '/'), trim($dir, '/'));
+    }
+
+    /**
+     * 创建文件夹
+     *
+     * @param $dir
+     */
+    public function makeDirectory($dir)
+    {
+        if (!file_exists($dir)) @mkdir($dir, 0777, true);
+    }
+
+    /**
+     * 上传媒体图片
+     *
+     * @param UploadedFile $file
+     * @param string $dir
+     * @param string $filename 文件夹名称
+     */
+    public function media(UploadedFile $file, string $dir, string $filename)
+    {
+        $save_dir = $this->driver($dir);
+        $this->makeDirectory($dir);
+
+        $file->moveTo(sprintf('%s/%s', $save_dir, $filename));
+        return $file->isMoved() ? sprintf('/%s/%s', trim($dir, '/'), $filename) : false;
+    }
+}

app/helper.php

@@ -151,7 +151,7 @@ function diff_date($day1, $day2)
  */
 function get_media_url(string $path)
 {
-    return '/' . $path;
+    return config('domain.img_url') . $path;
 }
 
 /**

config/autoload/server.php

@@ -46,7 +46,7 @@ return [
     ],
     'settings' => [
         'enable_coroutine' => true,
-        'worker_num' => swoole_cpu_num(),
+        'worker_num' => 1,
         'pid_file' => BASE_PATH . '/runtime/hyperf.pid',
         'open_tcp_nodelay' => true,
         'max_coroutine' => 100000,

config/config.php

@@ -1,6 +1,7 @@
 <?php
 
 declare(strict_types=1);
+
 /**
  * This file is part of Hyperf.
  *
@@ -9,6 +10,7 @@ declare(strict_types=1);
  * @contact  group@hyperf.io
  * @license  https://github.com/hyperf/hyperf/blob/master/LICENSE
  */
+
 use Hyperf\Contract\StdoutLoggerInterface;
 use Psr\Log\LogLevel;
 
@@ -17,10 +19,18 @@ return [
     'app_env' => env('APP_ENV', 'dev'),
     'scan_cacheable' => env('SCAN_CACHEABLE', false),
 
-    'ip_address'=>env('IP_ADDRESS', ''),
+    'ip_address' => env('IP_ADDRESS', ''),
 
     // 运行模式
-    'run_mode'=>'cluster',
+    'run_mode' => 'cluster',
+
+    // 域名相关配置
+    'domain' => [
+        'web_url' => env('WEB_URL', ''),//Web 端首页地址
+        'img_url' => env('IMG_URL', ''),//设置文件图片访问的域名
+    ],
+    'upload_dir'=>env('UPLOAD_PATH',''),
+
     StdoutLoggerInterface::class => [
         'log_level' => [
             LogLevel::ALERT,