import com.notnoop.apns.ApnsNotification; //导入依赖的package包/类
public ApnsConnectionImpl(SocketFactory factory, String host,
        int port, Proxy proxy,
        ReconnectPolicy reconnectPolicy, ApnsDelegate delegate,
        boolean errorDetection, int cacheLength, boolean autoAdjustCacheLength) {
    this.factory = factory;
    this.host = host;
    this.port = port;
    this.reconnectPolicy = reconnectPolicy;
    this.delegate = delegate == null ? ApnsDelegate.EMPTY : delegate;
    this.proxy = proxy;
    this.errorDetection = errorDetection;
    this.cacheLength = cacheLength;
    this.autoAdjustCacheLength = autoAdjustCacheLength;
    cachedNotifications = new ConcurrentLinkedQueue<ApnsNotification>();
    notificationsBuffer = new ConcurrentLinkedQueue<ApnsNotification>();
}
 

import com.notnoop.apns.ApnsNotification; //导入依赖的package包/类
@Override
public void push(ApnsNotification message) throws NetworkIOException {
	if (batch.isEmpty()) {
		firstMessageArrivedTime = System.nanoTime();
	}
	
	long sincFirstMessageSec = (System.nanoTime() - firstMessageArrivedTime) / 1000 / 1000 / 1000;
	
	if (taskFuture != null && sincFirstMessageSec < maxBatchWaitTimeInSec) {
		taskFuture.cancel(false);
	}
	
	batch.add(message);
	
	if (taskFuture == null || taskFuture.isDone()) {
		taskFuture = scheduleService.schedule(batchRunner, batchWaitTimeInSec, TimeUnit.SECONDS);
	}
}
 

import com.notnoop.apns.ApnsNotification; //导入依赖的package包/类
/**
 * Push simple (text only) message to client.
 * 
 * @param roleName
 * @param deviceToken
 * @param text
 * @return
 */
public final boolean pushSimpleMessage(final String roleName, 
		final String deviceToken, final String text, int expireSeconds) {
	boolean success = true;
	if ( StringUtil.checkNotEmpty(deviceToken) && iosApnService != null ) {
		GameContext.getInstance().runSmallTask(new Runnable() {
			public void run() {
				//ios device
				PayloadBuilder builder = PayloadBuilder.newPayload();
				builder.badge(1);
				builder.alertBody(text);
				ApnsNotification notif = iosApnService.push(deviceToken, builder.build());
			}
		});
		logger.debug("Push message '{}' to ios device", text);
	} else {
		//android devices
		String offlineBox = getOfflineboxName(roleName);
		pushValueInList(offlineBox, text, 10, expireSeconds);
		logger.debug("Push message '{}' to android device", text);
	}
	return success;
}
 

import com.notnoop.apns.ApnsNotification; //导入依赖的package包/类
public void start() {
    if (started.getAndSet(true)) {
        // I prefer if we throw a runtime IllegalStateException here,
        // but I want to maintain semantic backward compatibility.
        // So it is returning immediately here
        return;
    }

    service.start();
    shouldContinue = true;
    thread = new Thread() {
        public void run() {
            while (shouldContinue) {
                try {
                    ApnsNotification msg = queue.take();
                    service.push(msg);
                } catch (InterruptedException e) {
                }
            }
        }
    };
    thread.start();
}
 

import com.notnoop.apns.ApnsNotification; //导入依赖的package包/类
public ApnsConnectionImpl(SocketFactory factory, String host, int port, Proxy proxy, String proxyUsername, String proxyPassword,
                          ReconnectPolicy reconnectPolicy, ApnsDelegate delegate, boolean errorDetection, ThreadFactory tf, int cacheLength,
                          boolean autoAdjustCacheLength, int readTimeout, int connectTimeout) {
    this.factory = factory;
    this.host = host;
    this.port = port;
    this.reconnectPolicy = reconnectPolicy;
    this.delegate = delegate == null ? ApnsDelegate.EMPTY : delegate;
    this.proxy = proxy;
    this.errorDetection = errorDetection;
    this.threadFactory = tf == null ? defaultThreadFactory() : tf;
    this.cacheLength = cacheLength;
    this.autoAdjustCacheLength = autoAdjustCacheLength;
    this.readTimeout = readTimeout;
    this.connectTimeout = connectTimeout;
    this.proxyUsername = proxyUsername;
    this.proxyPassword = proxyPassword;
    cachedNotifications = new ConcurrentLinkedQueue<ApnsNotification>();
    notificationsBuffer = new ConcurrentLinkedQueue<ApnsNotification>();
}
 

import com.notnoop.apns.ApnsNotification; //导入依赖的package包/类
public void sendMessage(final ApnsNotification m) throws NetworkIOException {
    Future<Void> future = executors.submit(new Callable<Void>() {
        public Void call() throws Exception {
            uniquePrototype.get().sendMessage(m);
            return null;
        }
    });
    try {
        future.get();
    } catch (InterruptedException ie) {
        Thread.currentThread().interrupt();
    } catch (ExecutionException ee) {
        if (ee.getCause() instanceof NetworkIOException) {
            throw (NetworkIOException) ee.getCause();
        }
    }
}
 

import com.notnoop.apns.ApnsNotification; //导入依赖的package包/类
@Override
public void push(ApnsNotification message) throws NetworkIOException {
	if (batch.isEmpty()) {
		firstMessageArrivedTime = System.nanoTime();
	}
	
	long sinceFirstMessageSec = (System.nanoTime() - firstMessageArrivedTime) / 1000 / 1000 / 1000;
	
	if (taskFuture != null && sinceFirstMessageSec < maxBatchWaitTimeInSec) {
		taskFuture.cancel(false);
	}
	
	batch.add(message);
	
	if (taskFuture == null || taskFuture.isDone()) {
		taskFuture = scheduleService.schedule(batchRunner, batchWaitTimeInSec, TimeUnit.SECONDS);
	}
}
 

import com.notnoop.apns.ApnsNotification; //导入依赖的package包/类
@Test
public void simpleBatchWait_one() throws IOException, InterruptedException {
	// send message
	ApnsNotification message = service.push("1234", "{}");

	// make sure no message was send yet
	verify(prototype, times(0)).copy();
	verify(prototype, times(0)).sendMessage(message);
	verify(prototype, times(0)).close();

	Thread.sleep(delayTimeInSec_millis + /* for sure */250);

	// verify batch sends and close the connection
	verify(prototype, times(1)).copy();
	verify(prototype, times(1)).sendMessage(message);
	verify(prototype, times(1)).close();
}
 

import com.notnoop.apns.ApnsNotification; //导入依赖的package包/类
private void notify(Exchange exchange) throws ApnsException {
    MessageType messageType = getHeaderMessageType(exchange, MessageType.STRING);

    if (messageType == MessageType.APNS_NOTIFICATION) {
        ApnsNotification apnsNotification = exchange.getIn().getBody(ApnsNotification.class);
        getEndpoint().getApnsService().push(apnsNotification);
    } else {
        constructNotificationAndNotify(exchange, messageType);
    }
}
 

import com.notnoop.apns.ApnsNotification; //导入依赖的package包/类
public synchronized void sendMessage(ApnsNotification m, boolean fromBuffer) throws NetworkIOException {

        int attempts = 0;
        while (true) {
            try {
                attempts++;
                Socket socket = socket();
                socket.getOutputStream().write(m.marshall());
                socket.getOutputStream().flush();
                cacheNotification(m);

                delegate.messageSent(m, fromBuffer);

                logger.debug("Message \"{}\" sent", m);

                attempts = 0;
                drainBuffer();
                break;
            } catch (Exception e) {
                Utilities.close(socket);
                socket = null;
                if (attempts >= RETRIES) {
                    logger.error("Couldn't send message after " + RETRIES + " retries." + m, e);
                    delegate.messageSendFailed(m, e);
                    Utilities.wrapAndThrowAsRuntimeException(e);
                }
                // The first failure might be due to closed connection
                // don't delay quite yet
                if (attempts != 1) {
                    // Do not spam the log files when the APNS server closed the socket (due to a
                    // bad token, for example), only log when on the second retry.
                    logger.info("Failed to send message " + m + "... trying again after delay", e);
                    Utilities.sleep(DELAY_IN_MS);
                }
            }
        }
    }
 

import com.notnoop.apns.ApnsNotification; //导入依赖的package包/类
private void cacheNotification(ApnsNotification notification) {
    cachedNotifications.add(notification);
    while (cachedNotifications.size() > cacheLength) {
        cachedNotifications.poll();
        logger.debug("Removing notification from cache " + notification);
    }
}
 

import com.notnoop.apns.ApnsNotification; //导入依赖的package包/类
public void sendMessage(final ApnsNotification m) throws NetworkIOException {
    executors.execute(new Runnable() {
        public void run() {
            uniquePrototype.get().sendMessage(m);
        }
    });
}
 

import com.notnoop.apns.ApnsNotification; //导入依赖的package包/类
public void run() {
	ApnsConnection newConnection = prototype.copy();
	try {
		ApnsNotification msg = null;
		while ((msg = batch.poll()) != null) {
			try {
				newConnection.sendMessage(msg);
			} catch (NetworkIOException e) {
				continue;
			}
		}
	} finally {
		Utilities.close(newConnection);
	}
}
 

import com.notnoop.apns.ApnsNotification; //导入依赖的package包/类
@Override
public void push(ApnsNotification msg) {
    if (!started.get()) {
        throw new IllegalStateException("service hasn't be started or was closed");
    }
    queue.add(msg);
}
 

import com.notnoop.apns.ApnsNotification; //导入依赖的package包/类
public void start() {
    if (started.getAndSet(true)) {
        // I prefer if we throw a runtime IllegalStateException here,
        // but I want to maintain semantic backward compatibility.
        // So it is returning immediately here
        return;
    }

    service.start();
    shouldContinue = true;
    thread = new Thread() {
        public void run() {
            while (shouldContinue) {
                try {
                    ApnsNotification msg = queue.take();
                    service.push(msg);
                } catch (InterruptedException e) {
                	// ignore
                } catch (NetworkIOException e) {
                	// ignore: failed connect...
                } catch (Exception e) {
                	// weird if we reached here - something wrong is happening, but we shouldn't stop the service anyway!
                	logger.warn("Unexpected message caught... Shouldn't be here", e);
                }
            }
        }
    };
    thread.start();
}
 

import com.notnoop.apns.ApnsNotification; //导入依赖的package包/类
@Override
  public void messageSendFailed(ApnsNotification message, Throwable e) {
    LOGGER.error("APNS Message delivery failed for key:{}", key, e);
    if (message != null) {
      byte[] tokenByte = message.getDeviceToken();
      String deviceToken = Utilities.encodeHex(tokenByte);
      LOGGER.error("APNS Message delivery failed for key:{}", key, e);

      //TODO : Revisit failure handling. Commented on 12/17/2015
//      invalidateToken(deviceToken);
//      byte[] payload = message.getPayload();
//      String payloadString = fromBytes(payload);
//      if (payloadString != null) {
//        APNSPayloadInfo info = APNSPayloadInfo.parse(payloadString);
//        if (info != null && info.getMessageId() != null) {
//          Integer errorCode = null;
//          if (e instanceof ApnsDeliveryErrorException) {
//            errorCode = Integer.valueOf(((ApnsDeliveryErrorException) e).getDeliveryError().code());
//          }
//          //mark push message state as error
//          updatePushMessageState(info.getMessageId(), errorCode);
//        }
//      }
    } else {
      LOGGER.error("APNS Message delivery failed but ApnsNotification message is null and hence can't invalidate the token");
    }
  }
 

import com.notnoop.apns.ApnsNotification; //导入依赖的package包/类
public synchronized void sendMessage(ApnsNotification m) throws NetworkIOException {
    int attempts = 0;
    while (true) {
        try {
            attempts++;
            Socket socket = socket();
            socket.getOutputStream().write(m.marshall());
            socket.getOutputStream().flush();
            delegate.messageSent(m);

            logger.debug("Message \"{}\" sent", m);

            attempts = 0;
            break;
        } catch (Exception e) {
            if (attempts >= RETRIES) {
                logger.error("Couldn't send message " + m, e);
                delegate.messageSendFailed(m, e);
                Utilities.wrapAndThrowAsRuntimeException(e);
            }
            logger.warn("Failed to send message " + m + "... trying again", e);
            // The first failure might be due to closed connection
            // don't delay quite yet
            if (attempts != 1)
                Utilities.sleep(DELAY_IN_MS);
            Utilities.close(socket);
            socket = null;
        }
    }
}
 

import com.notnoop.apns.ApnsNotification; //导入依赖的package包/类
private synchronized void sendMessage(ApnsNotification m, boolean fromBuffer) throws NetworkIOException {
    logger.debug("sendMessage {} fromBuffer: {}", m, fromBuffer);

    int attempts = 0;
    while (true) {
        try {
            attempts++;
            Socket socket = getOrCreateSocket();
            socket.getOutputStream().write(m.marshall());
            socket.getOutputStream().flush();
            cacheNotification(m);

            delegate.messageSent(m, fromBuffer);

            //logger.debug("Message \"{}\" sent", m);
            attempts = 0;
            break;
        } catch (IOException e) {
            Utilities.close(socket);
            if (attempts >= RETRIES) {
                logger.error("Couldn't send message after " + RETRIES + " retries." + m, e);
                delegate.messageSendFailed(m, e);
                Utilities.wrapAndThrowAsRuntimeException(e);
            }
            // The first failure might be due to closed connection (which in turn might be caused by
            // a message containing a bad token), so don't delay for the first retry.
            //
            // Additionally we don't want to spam the log file in this case, only after the second retry
            // which uses the delay.

            if (attempts != 1) {
                logger.info("Failed to send message " + m + "... trying again after delay", e);
                Utilities.sleep(DELAY_IN_MS);
            }
        }
    }
}
 

import com.notnoop.apns.ApnsNotification; //导入依赖的package包/类
public void run() {
	ApnsConnection newConnection = prototype.copy();
	try {
		ApnsNotification msg;
		while ((msg = batch.poll()) != null) {
			try {
				newConnection.sendMessage(msg);
			} catch (NetworkIOException e) {
                      logger.warn("Network exception sending message msg "+ msg.getIdentifier(), e);
                  }
		}
	} finally {
		Utilities.close(newConnection);
	}
}
 

import com.notnoop.apns.ApnsNotification; //导入依赖的package包/类
public QueuedApnsService(ApnsService service, final ThreadFactory tf) {
    super(null);
    this.service = service;
    this.queue = new LinkedBlockingQueue<ApnsNotification>();
    this.threadFactory = tf == null ? Executors.defaultThreadFactory() : tf;
    this.thread = null;
}