forked from mpusher/mpush
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
32 changed files
with
519 additions
and
134 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -25,7 +25,7 @@ | |
| * @author [email protected] (夜色) | ||
| */ | ||
| public interface AckCallback { | ||
| void onSuccess(AckContext context); | ||
| void onSuccess(AckTask context); | ||
|
|
||
| void onTimeout(AckContext context); | ||
| void onTimeout(AckTask context); | ||
| } | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -19,48 +19,58 @@ | |
|
|
||
| package com.mpush.core.ack; | ||
|
|
||
| import com.mpush.api.Message; | ||
|
|
||
| import java.util.concurrent.Future; | ||
| import java.util.concurrent.ScheduledExecutorService; | ||
|
|
||
| /** | ||
| * Created by ohun on 16/9/5. | ||
| * | ||
| * @author [email protected] (夜色) | ||
| */ | ||
| public final class AckContext implements Runnable { | ||
| private AckCallback callback; | ||
| public final class AckTask implements Runnable { | ||
| final int ackMessageId; | ||
| private final Message message; | ||
|
|
||
| private int sessionId; | ||
| private Future<?> future; | ||
| private AckCallback callback; | ||
| private Future<?> timeoutFuture; | ||
|
|
||
| public AckContext() { | ||
| public AckTask(Message message, int ackMessageId) { | ||
| this.message = message; | ||
| this.ackMessageId = ackMessageId; | ||
| } | ||
|
|
||
| public void setSessionId(int sessionId) { | ||
| this.sessionId = sessionId; | ||
| public static AckTask from(Message message, int ackMessageId) { | ||
| return new AckTask(message, ackMessageId); | ||
| } | ||
|
|
||
| public void setFuture(Future<?> future) { | ||
| this.future = future; | ||
| this.timeoutFuture = future; | ||
| } | ||
|
|
||
| public ScheduledExecutorService getExecutor() { | ||
| return message.getConnection().getChannel().eventLoop(); | ||
| } | ||
|
|
||
| public AckContext setCallback(AckCallback callback) { | ||
| public AckTask setCallback(AckCallback callback) { | ||
| this.callback = callback; | ||
| return this; | ||
| } | ||
|
|
||
| private boolean tryDone() { | ||
| return future.cancel(true); | ||
| return timeoutFuture.cancel(true); | ||
| } | ||
|
|
||
| public void success() { | ||
| public void onResponse() { | ||
| if (tryDone()) { | ||
| callback.onSuccess(this); | ||
| callback = null; | ||
| } | ||
| } | ||
|
|
||
| public void timeout() { | ||
| AckContext context = AckMessageQueue.I.getAndRemove(sessionId); | ||
| public void onTimeout() { | ||
| AckTask context = AckTaskQueue.I.getAndRemove(ackMessageId); | ||
| if (context != null && tryDone()) { | ||
| callback.onTimeout(this); | ||
| callback = null; | ||
|
|
@@ -69,13 +79,13 @@ public void timeout() { | |
|
|
||
| @Override | ||
| public String toString() { | ||
| return "AckContext{" + | ||
| ", sessionId=" + sessionId + | ||
| return "{" + | ||
| ", ackMessageId=" + ackMessageId + | ||
| '}'; | ||
| } | ||
|
|
||
| @Override | ||
| public void run() { | ||
| timeout(); | ||
| onTimeout(); | ||
| } | ||
| } | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -35,28 +35,31 @@ | |
| * | ||
| * @author [email protected] (夜色) | ||
| */ | ||
| public final class AckMessageQueue extends BaseService { | ||
| private final Logger logger = LoggerFactory.getLogger(AckMessageQueue.class); | ||
| public final class AckTaskQueue extends BaseService { | ||
| private final Logger logger = LoggerFactory.getLogger(AckTaskQueue.class); | ||
|
|
||
| private static final int DEFAULT_TIMEOUT = 3000; | ||
| public static final AckMessageQueue I = new AckMessageQueue(); | ||
| public static final AckTaskQueue I = new AckTaskQueue(); | ||
|
|
||
| private final ConcurrentMap<Integer, AckContext> queue = new ConcurrentHashMap<>(); | ||
| private final ConcurrentMap<Integer, AckTask> queue = new ConcurrentHashMap<>(); | ||
| private ScheduledExecutorService scheduledExecutor; | ||
|
|
||
| private AckMessageQueue() { | ||
| private AckTaskQueue() { | ||
| } | ||
|
|
||
| public void add(int sessionId, AckContext context, int timeout) { | ||
| queue.put(sessionId, context); | ||
| context.setSessionId(sessionId); | ||
| context.setFuture(scheduledExecutor.schedule(context, | ||
| public void add(AckTask task, int timeout) { | ||
| queue.put(task.ackMessageId, task); | ||
|
|
||
| //使用 task.getExecutor() 并没更快 | ||
| task.setFuture(scheduledExecutor.schedule(task, | ||
| timeout > 0 ? timeout : DEFAULT_TIMEOUT, | ||
| TimeUnit.MILLISECONDS | ||
| )); | ||
|
|
||
| logger.debug("one ack task add to queue, task={}, timeout={}", task, timeout); | ||
| } | ||
|
|
||
| public AckContext getAndRemove(int sessionId) { | ||
| public AckTask getAndRemove(int sessionId) { | ||
| return queue.remove(sessionId); | ||
| } | ||
|
|
||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.