diff --git a/channel/message-queue.go b/channel/message-queue.go
new file mode 100644
index 0000000..b738d39
--- /dev/null
+++ b/channel/message-queue.go
@@ -0,0 +1,63 @@
+package channel
+
+import (
+ "message-pusher/common"
+ "message-pusher/model"
+)
+
+var AsyncMessageQueue chan int
+var AsyncMessageQueueSize = 128
+var AsyncMessageSenderNum = 2
+
+func init() {
+ AsyncMessageQueue = make(chan int, AsyncMessageQueueSize)
+ for i := 0; i < AsyncMessageSenderNum; i++ {
+ go asyncMessageSender()
+ }
+}
+
+// LoadAsyncMessages loads async pending messages from database.
+// We have to wait the database connection is ready.
+func LoadAsyncMessages() {
+ ids, err := model.GetAsyncPendingMessageIds()
+ if err != nil {
+ common.FatalLog("failed to load async pending messages: " + err.Error())
+ }
+ for _, id := range ids {
+ AsyncMessageQueue <- id
+ }
+}
+
+func asyncMessageSenderHelper(message *model.Message) error {
+ user, err := model.GetUserById(message.UserId, false)
+ if err != nil {
+ return err
+ }
+ channel_, err := model.GetChannelByName(message.Channel, user.Id)
+ if err != nil {
+ return err
+ }
+ return SendMessage(message, user, channel_)
+}
+
+func asyncMessageSender() {
+ for {
+ id := <-AsyncMessageQueue
+ message, err := model.GetMessageById(id)
+ if err != nil {
+ common.SysError("async message sender error: " + err.Error())
+ continue
+ }
+ err = asyncMessageSenderHelper(message)
+ status := common.MessageSendStatusFailed
+ if err != nil {
+ common.SysError("async message sender error: " + err.Error())
+ } else {
+ status = common.MessageSendStatusSent
+ }
+ err = message.UpdateStatus(status)
+ if err != nil {
+ common.SysError("async message sender error: " + err.Error())
+ }
+ }
+}
diff --git a/common/constants.go b/common/constants.go
index cc8fd5f..01a00ee 100644
--- a/common/constants.go
+++ b/common/constants.go
@@ -101,10 +101,11 @@ const (
)
const (
- MessageSendStatusUnknown = 0
- MessageSendStatusPending = 1
- MessageSendStatusSent = 2
- MessageSendStatusFailed = 3
+ MessageSendStatusUnknown = 0
+ MessageSendStatusPending = 1
+ MessageSendStatusSent = 2
+ MessageSendStatusFailed = 3
+ MessageSendStatusAsyncPending = 4
)
const (
diff --git a/controller/message.go b/controller/message.go
index 19a638b..621b14e 100644
--- a/controller/message.go
+++ b/controller/message.go
@@ -38,6 +38,7 @@ func GetPushMessage(c *gin.Context) {
Desp: c.Query("desp"),
Short: c.Query("short"),
OpenId: c.Query("openid"),
+ Async: c.Query("async") == "true",
}
keepCompatible(&message)
pushMessageHelper(c, &message)
@@ -55,6 +56,7 @@ func PostPushMessage(c *gin.Context) {
Desp: c.PostForm("desp"),
Short: c.PostForm("short"),
OpenId: c.PostForm("openid"),
+ Async: c.PostForm("async") == "true",
}
if message == (model.Message{}) {
// Looks like the user is using JSON
@@ -142,6 +144,7 @@ func pushMessageHelper(c *gin.Context, message *model.Message) {
c.JSON(http.StatusOK, gin.H{
"success": true,
"message": "",
+ "uuid": message.Link,
})
}
@@ -149,6 +152,7 @@ func saveAndSendMessage(user *model.User, message *model.Message, channel_ *mode
if channel_.Status != common.ChannelStatusEnabled {
return errors.New("该渠道已被禁用")
}
+ common.MessageCount += 1 // We don't need to use atomic here because it's not a critical value
message.Link = common.GetUUID()
if message.URL == "" {
message.URL = fmt.Sprintf("%s/message/%s", common.ServerAddress, message.Link)
@@ -158,25 +162,36 @@ func saveAndSendMessage(user *model.User, message *model.Message, channel_ *mode
defer func() {
// Update the status of the message
status := common.MessageSendStatusFailed
- if success {
- status = common.MessageSendStatusSent
+ if message.Async {
+ status = common.MessageSendStatusAsyncPending
+ } else {
+ if success {
+ status = common.MessageSendStatusSent
+ }
}
err := message.UpdateStatus(status)
if err != nil {
common.SysError("failed to update the status of the message: " + err.Error())
}
+ if message.Async {
+ channel.AsyncMessageQueue <- message.Id
+ }
}()
err := message.UpdateAndInsert(user.Id)
if err != nil {
return err
}
} else {
+ if message.Async {
+ return errors.New("异步发送消息需要用户具备消息持久化的权限")
+ }
message.Link = "unsaved" // This is for user to identify whether the message is saved
}
- err := channel.SendMessage(message, user, channel_)
- common.MessageCount += 1 // We don't need to use atomic here because it's not a critical value
- if err != nil {
- return err
+ if !message.Async {
+ err := channel.SendMessage(message, user, channel_)
+ if err != nil {
+ return err
+ }
}
success = true
return nil // After this line, the message status will be updated
@@ -258,7 +273,7 @@ func GetUserMessages(c *gin.Context) {
func GetMessage(c *gin.Context) {
messageId, _ := strconv.Atoi(c.Param("id"))
userId := c.GetInt("id")
- message, err := model.GetMessageById(messageId, userId)
+ message, err := model.GetMessageByIds(messageId, userId)
if err != nil {
c.JSON(http.StatusOK, gin.H{
"success": false,
@@ -296,7 +311,7 @@ func ResendMessage(c *gin.Context) {
messageId, _ := strconv.Atoi(c.Param("id"))
userId := c.GetInt("id")
helper := func() error {
- message, err := model.GetMessageById(messageId, userId)
+ message, err := model.GetMessageByIds(messageId, userId)
message.Id = 0
if err != nil {
return err
diff --git a/main.go b/main.go
index 7907ac6..53a9144 100644
--- a/main.go
+++ b/main.go
@@ -33,6 +33,7 @@ func main() {
if err != nil {
common.FatalLog(err)
}
+ go channel.LoadAsyncMessages()
defer func() {
err := model.CloseDB()
if err != nil {
diff --git a/model/message.go b/model/message.go
index c7cf9b9..dea22e0 100644
--- a/model/message.go
+++ b/model/message.go
@@ -18,14 +18,15 @@ type Message struct {
HTMLContent string `json:"html_content" gorm:"-:all"`
Timestamp int64 `json:"timestamp" gorm:"type:bigint"`
Link string `json:"link" gorm:"unique;index"`
- To string `json:"to" gorm:"column:to"` // if specified, will send to this user(s)
- Status int `json:"status" gorm:"default:0"` // pending, sent, failed
- OpenId string `json:"openid" gorm:"-:all"` // alias for to
- Desp string `json:"desp" gorm:"-:all"` // alias for content
- Short string `json:"short" gorm:"-:all"` // alias for description
+ To string `json:"to" gorm:"column:to"` // if specified, will send to this user(s)
+ Status int `json:"status" gorm:"default:0;index"` // pending, sent, failed
+ OpenId string `json:"openid" gorm:"-:all"` // alias for to
+ Desp string `json:"desp" gorm:"-:all"` // alias for content
+ Short string `json:"short" gorm:"-:all"` // alias for description
+ Async bool `json:"async" gorm:"-"` // if true, will send message asynchronously
}
-func GetMessageById(id int, userId int) (*Message, error) {
+func GetMessageByIds(id int, userId int) (*Message, error) {
if id == 0 || userId == 0 {
return nil, errors.New("id 或 userId 为空！")
}
@@ -34,6 +35,20 @@ func GetMessageById(id int, userId int) (*Message, error) {
return &message, err
}
+func GetMessageById(id int) (*Message, error) {
+ if id == 0 {
+ return nil, errors.New("id 为空！")
+ }
+ message := Message{Id: id}
+ err := DB.Where(message).First(&message).Error
+ return &message, err
+}
+
+func GetAsyncPendingMessageIds() (ids []int, err error) {
+ err = DB.Model(&Message{}).Where("status = ?", common.MessageSendStatusAsyncPending).Pluck("id", &ids).Error
+ return ids, err
+}
+
func GetMessageByLink(link string) (*Message, error) {
if link == "" {
return nil, errors.New("link 为空！")
diff --git a/web/src/components/MessagesTable.js b/web/src/components/MessagesTable.js
index 7ab584f..8f8dbbd 100644
--- a/web/src/components/MessagesTable.js
+++ b/web/src/components/MessagesTable.js
@@ -17,7 +17,7 @@ function renderStatus(status) {
case 1:
return (
);
case 2:
@@ -32,6 +32,12 @@ function renderStatus(status) {
发送失败
);
+ case 4:
+ return (
+
+ );
default:
return (