retweet wait 5m

api_service.go

@@ -88,7 +88,7 @@ func Project(c *fiber.Ctx) error {
 
 	}
 
-	cli := NewClient(req.Config)
+	cli := NewClient(req.Config, nil)
 
 	me, err := cli.Me()
 
@@ -264,6 +264,13 @@ func TaskStop(c *fiber.Ctx) error {
 		})
 	}
 
+	if err := Worker.StopJob(req.User, req.TaskType); err != nil {
+		return c.JSON(Res{
+			Code: 500,
+			Msg:  err.Error(),
+		})
+	}
+
 	err := StopTaskUpdate(req)
 
 	if err != nil {

client.go

@@ -38,7 +38,7 @@ PER USER
 */
 
 var LikeRateLimit *rate.Limiter = rate.NewLimiter(rate.Every(3*time.Minute), 1)
-var RetweetRateLimit *rate.Limiter = rate.NewLimiter(rate.Every(3*time.Minute), 1)
+var RetweetRateLimit *rate.Limiter = rate.NewLimiter(rate.Every(5*time.Minute), 1)
 
 type authorize struct {
 	Token string
@@ -65,17 +65,17 @@ type Config struct {
 	Token             string `json:"token"`
 }
 
-// func NewLikeClient(cfg Config) *Client {
-// 	return NewClient(cfg, LikeRateLimit)
-// }
+func NewLikeClient(cfg Config) *Client {
+	return NewClient(cfg, LikeRateLimit)
+}
 
-// func NewRetweeterClient(cfg Config) *Client {
-// 	return NewClient(cfg, RetweetRateLimit)
-// }
+func NewRetweeterClient(cfg Config) *Client {
+	return NewClient(cfg, RetweetRateLimit)
+}
 
-//func NewClient(cfg Config, rts *rate.Limiter) *Client {
+func NewClient(cfg Config, rts *rate.Limiter) *Client {
 
-func NewClient(cfg Config) *Client {
+	//func NewClient(cfg Config) *Client {
 
 	//twitterAPIKey := "lVnj6Ox9HPcI4LwArSSYU7Pba"                                //os.Getenv("TWITTER_API_KEY")
 	//twitterAPIKeySecret := "QMSnWG4QwyXWBVW2hQazzxhw9cSjd32CDfXGkg2DEaUUdscCRZ" //os.Getenv("TWITTER_API_KEY_SECRET")
@@ -100,10 +100,10 @@ func NewClient(cfg Config) *Client {
 	}
 
 	return &Client{
-		Client: twitterClient,
-		// Ratelimiter:           rts,
-		RetweeterRatelimiter:  RetweetRateLimit,
-		LikingUserRatelimiter: LikeRateLimit,
+		Client:      twitterClient,
+		Ratelimiter: rts,
+		//RetweeterRatelimiter:  RetweetRateLimit,
+		//LikingUserRatelimiter: LikeRateLimit,
 	}
 
 }
@@ -142,9 +142,9 @@ func (c *Client) Retweeters(tweetId string, next string) ([]*twitter.UserObj, st
 	// TODO: Fix performance by removing unneeded allocaton here
 
 	ctx := context.Background()
-	//err := c.Ratelimiter.Wait(ctx) // This is a blocking call.
+	err := c.Ratelimiter.Wait(ctx) // This is a blocking call.
 
-	err := c.RetweeterRatelimiter.Wait(ctx) // This is a blocking call.
+	//err := c.RetweeterRatelimiter.Wait(ctx) // This is a blocking call.
 	if err != nil {
 		return nil, "", nil, err
 	}
@@ -169,8 +169,8 @@ func (c *Client) TweetLikingUsers(tweetId string, next string) ([]*twitter.UserO
 	// TODO: Fix performance by removing unneeded allocaton here
 
 	ctx := context.Background()
-	//err := c.Ratelimiter.Wait(ctx) // This is a blocking call.
-	err := c.LikingUserRatelimiter.Wait(ctx) // This is a blocking call.
+	err := c.Ratelimiter.Wait(ctx) // This is a blocking call.
+	//err := c.LikingUserRatelimiter.Wait(ctx) // This is a blocking call.
 	if err != nil {
 		return nil, "", nil, err
 	}
@@ -192,7 +192,7 @@ func (c *Client) TweetLikingUsers(tweetId string, next string) ([]*twitter.UserO
 
 }
 
-func (c *Client) Me() (map[string]*twitter.UserDictionary, error) {
+func (c *Client) Usage() (map[string]*twitter.UserDictionary, error) {
 
 	// ctx is generated here only to use with Ratelimiter
 	// TODO: Fix performance by removing unneeded allocaton here

docs/swagger.yaml

@@ -358,7 +358,7 @@ components:
           example: "1800805503066661056"
     StopTaskReq:
       required:
-        - task_id
+        - user_id
         - task_type
       type: object
       properties:
@@ -374,7 +374,6 @@ components:
           type: string
           example: "1800805503066661056"
         user_id:
-          description: 方便识别用户身份信息
           type: string
           example: "1783145144700874752" 
         # start:

task.go

@@ -9,19 +9,35 @@ import (
 type Work struct {
 	Lock sync.Mutex
 
-	Task       map[string]chan<- interface{}
-	userClient map[string]*Client
+	Task map[string]chan<- interface{}
+	// userClient map[string]*Client
 }
 
 var Worker Work
 
 func init() {
 	Worker = Work{
-		Task:       make(map[string]chan<- interface{}),
-		userClient: make(map[string]*Client),
+		Task: make(map[string]chan<- interface{}),
+		//userClient: make(map[string]*Client),
 	}
 }
 
+func (w *Work) StopJob(userId, taskType string) error {
+	w.Lock.Lock()
+
+	if v, ok := w.Task[userId+"-"+taskType]; ok {
+		close(v)
+	} else {
+		return fmt.Errorf("%s do not run", userId+"-"+taskType)
+	}
+
+	// done := w.RunJob(t)
+	// w.Task[t.UserId+"-"+t.TaskType] = done
+	w.Lock.Unlock()
+
+	return nil
+}
+
 func (w *Work) AddJob(t TaskJob) error {
 	w.Lock.Lock()
 
@@ -44,11 +60,10 @@ func (w *Work) RunJob(t TaskJob) chan<- interface{} {
 	go func() {
 		var cli *Client
 
-		if v, ok := w.userClient[t.UserId]; ok {
-			cli = v
+		if t.TaskType == RetweetType {
+			cli = NewRetweeterClient(t.Config)
 		} else {
-			cli = NewClient(t.Config)
-			w.userClient[t.UserId] = cli
+			cli = NewLikeClient(t.Config)
 		}
 
 		page := NewPageUsers(NewIdx(t.Idx))