		void amqp() throws Throwable {
			final PushQueue dev_queue = Push.queue("push.p12", "pusubi", false,
					1);
			dev_queue.start();
			final PushQueue prod_queue = Push.queue("pushprod.p12", "pusubi",
					true, 1);
			prod_queue.start();
			for (;;) {
				ConnectionFactory connectionFactory = new ConnectionFactory();
				connectionFactory.setHost("bumblebee.musubi.us");
				connectionFactory.setConnectionTimeout(30 * 1000);
				connectionFactory.setRequestedHeartbeat(30);
				Connection connection = connectionFactory.newConnection();
				mIncomingChannel = connection.createChannel();
				mConsumer = new DefaultConsumer(mIncomingChannel) {
					@Override
					public void handleDelivery(final String consumerTag,
							final Envelope envelope,
							final BasicProperties properties, final byte[] body)
							throws IOException {
						HashSet<String> threadDevices = new HashSet<String>();
						synchronized (mNotifiers) {
							String identity = mConsumers.get(consumerTag);
							if (identity == null)
								return;
							HashSet<String> devices = mNotifiers.get(identity);
							if (devices == null)
								return;
							threadDevices.addAll(devices);
						}
						Message m = null;
						try {
							m = mMapper.readValue(body, Message.class);
						} catch (IOException e) {
							new RuntimeException(
									"Failed to parse BSON of outer message", e)
									.printStackTrace();
							return;
						}
						if (m.l)
							return;
						String sender_exchange;
						try {
							sender_exchange = encodeAMQPname("ibeidentity-",
									new IBHashedIdentity(m.s.i).at(0).identity_);
						} catch (CorruptIdentity e) {
							e.printStackTrace();
							return;
						}
						Date now = new Date();
						for (String device : threadDevices) {
							Listener l;
							synchronized (mNotifiers) {
								l = mListeners.get(device);
							}
							if (l.identityExchanges.contains(sender_exchange))
								continue;
							int new_value = 0;
							int amqp = 0;
							int local = 0;
							Date last;
							synchronized (mCounts) {
								BadgeData bd = mCounts.get(device);
								if (bd == null) {
									bd = new BadgeData();
									mCounts.put(device, bd);
								}
								bd.amqp++;
								amqp = bd.amqp;
								local = bd.local;
								new_value = bd.amqp + bd.local;
								last = bd.last;
								if (bd.last == null) {
									bd.last = now;
								} else if (bd.last != null
										&& now.getTime() - bd.last.getTime() > 3 * 60 * 1000) {
									bd.last = now;
									last = null;
								}
							}
							if (l.platform.equals("android")) {
								com.google.android.gcm.server.Message message = new com.google.android.gcm.server.Message.Builder()
										.addData("amqp", Integer.toString(amqp))
										.build();
								mGooglePushes.add(Pair.of(message, device));
							} else {
								try {
									boolean production = l.production != null
											&& l.production != false;
									PushNotificationPayload payload = PushNotificationPayload
											.complex();
									try {
										if (last == null) {
											payload.addAlert("New message");
											payload.addSound("default");
										}
										payload.addBadge(new_value);
										payload.addCustomDictionary("local",
												local);
										payload.addCustomDictionary("amqp",
												amqp);
									} catch (JSONException e) {
										e.printStackTrace();
										System.exit(1);
									}
									if (!production)
										dev_queue.add(payload, device);
									else
										prod_queue.add(payload, device);
								} catch (InvalidDeviceTokenFormatException e) {
									e.printStackTrace();
								}
							}
						}
					}
				};
				System.out.println("doing registrations");
				Set<String> notifiers = new HashSet<String>();
				synchronized (mNotifiers) {
					notifiers.addAll(mNotifiers.keySet());
				}
				for (String identity : notifiers) {
					DeclareOk x = mIncomingChannel.queueDeclare();
					System.out.println("listening " + identity);
					mIncomingChannel.exchangeDeclare(identity, "fanout", true);
					mIncomingChannel.queueBind(x.getQueue(), identity, "");
					String consumerTag = mIncomingChannel.basicConsume(
							x.getQueue(), true, mConsumer);
					synchronized (mNotifiers) {
						mQueues.put(identity, x.getQueue());
						mConsumers.put(consumerTag, identity);
					}
				}
				System.out.println("done registrations");
				long last = new Date().getTime();
				for (;;) {
					Runnable job = mJobs.poll(60, TimeUnit.SECONDS);
					long current = new Date().getTime();
					if (current - last > 60 * 1000) {
						PushedNotifications ps = dev_queue
								.getPushedNotifications();
						for (PushedNotification p : ps) {
							if (p.isSuccessful())
								continue;
							String invalidToken = p.getDevice().getToken();
							System.err.println("unregistering invalid token "
									+ invalidToken);
							unregister(invalidToken);
							Exception theProblem = p.getException();
							theProblem.printStackTrace();
							ResponsePacket theErrorResponse = p.getResponse();
							if (theErrorResponse != null) {
								System.out.println(theErrorResponse
										.getMessage());
							}
						}
						last = new Date().getTime();
						List<Device> inactiveDevices = Push.feedback(
								"push.p12", "pusubi", false);
						for (Device d : inactiveDevices) {
							String invalidToken = d.getToken();
							System.err
									.println("unregistering feedback failed token token "
											+ invalidToken);
							unregister(invalidToken);
						}
					}
					if (job == null)
						continue;
					job.run();
				}
			}
		}

		void amqp() throws Throwable {
			final PushQueue dev_queue = Push.queue("push.p12", "pusubi", false,
					1);
			dev_queue.start();
			final PushQueue prod_queue = Push.queue("pushprod.p12", "pusubi",
					true, 1);
			prod_queue.start();
			for (;;) {
				ConnectionFactory connectionFactory = new ConnectionFactory();
				connectionFactory.setHost("bumblebee.musubi.us");
				connectionFactory.setConnectionTimeout(30 * 1000);
				connectionFactory.setRequestedHeartbeat(30);
				Connection connection = connectionFactory.newConnection();
				mIncomingChannel = connection.createChannel();
				mConsumer = new DefaultConsumer(mIncomingChannel) {
					@Override
					public void handleDelivery(final String consumerTag,
							final Envelope envelope,
							final BasicProperties properties, final byte[] body)
							throws IOException {
						HashSet<String> threadDevices = new HashSet<String>();
						synchronized (mNotifiers) {
							String identity = mConsumers.get(consumerTag);
							if (identity == null)
								return;
							HashSet<String> devices = mNotifiers.get(identity);
							if (devices == null)
								return;
							threadDevices.addAll(devices);
						}
						Message m = null;
						try {
							m = mMapper.readValue(body, Message.class);
						} catch (IOException e) {
							new RuntimeException(
									"Failed to parse BSON of outer message", e)
									.printStackTrace();
							return;
						}
						if (m.l)
							return;
						String sender_exchange;
						try {
							sender_exchange = encodeAMQPname("ibeidentity-",
									new IBHashedIdentity(m.s.i).at(0).identity_);
						} catch (CorruptIdentity e) {
							e.printStackTrace();
							return;
						}
						Date now = new Date();
						for (String device : threadDevices) {
							Listener l;
							synchronized (mNotifiers) {
								l = mListeners.get(device);
							}
							if (l.identityExchanges.contains(sender_exchange))
								continue;
							int new_value = 0;
							int amqp = 0;
							int local = 0;
							Date last;
							synchronized (mCounts) {
								BadgeData bd = mCounts.get(device);
								if (bd == null) {
									bd = new BadgeData();
									mCounts.put(device, bd);
								}
								bd.amqp++;
								amqp = bd.amqp;
								local = bd.local;
								new_value = bd.amqp + bd.local;
								last = bd.last;
								if (bd.last == null) {
									bd.last = now;
								} else if (bd.last != null
										&& now.getTime() - bd.last.getTime() > 3 * 60 * 1000) {
									bd.last = now;
									last = null;
								}
							}
							if (l.platform != null && l.platform.equals("android")) {
								com.google.android.gcm.server.Message message = new com.google.android.gcm.server.Message.Builder()
										.addData("amqp", Integer.toString(amqp))
										.build();
								mGooglePushes.add(Pair.of(message, device));
							} else {
								try {
									boolean production = l.production != null
											&& l.production != false;
									PushNotificationPayload payload = PushNotificationPayload
											.complex();
									try {
										if (last == null) {
											payload.addAlert("New message");
											payload.addSound("default");
										}
										payload.addBadge(new_value);
										payload.addCustomDictionary("local",
												local);
										payload.addCustomDictionary("amqp",
												amqp);
									} catch (JSONException e) {
										e.printStackTrace();
										System.exit(1);
									}
									if (!production)
										dev_queue.add(payload, device);
									else
										prod_queue.add(payload, device);
								} catch (InvalidDeviceTokenFormatException e) {
									e.printStackTrace();
								}
							}
						}
					}
				};
				System.out.println("doing registrations");
				Set<String> notifiers = new HashSet<String>();
				synchronized (mNotifiers) {
					notifiers.addAll(mNotifiers.keySet());
				}
				for (String identity : notifiers) {
					DeclareOk x = mIncomingChannel.queueDeclare();
					System.out.println("listening " + identity);
					mIncomingChannel.exchangeDeclare(identity, "fanout", true);
					mIncomingChannel.queueBind(x.getQueue(), identity, "");
					String consumerTag = mIncomingChannel.basicConsume(
							x.getQueue(), true, mConsumer);
					synchronized (mNotifiers) {
						mQueues.put(identity, x.getQueue());
						mConsumers.put(consumerTag, identity);
					}
				}
				System.out.println("done registrations");
				long last = new Date().getTime();
				for (;;) {
					Runnable job = mJobs.poll(60, TimeUnit.SECONDS);
					long current = new Date().getTime();
					if (current - last > 60 * 1000) {
						PushedNotifications ps = dev_queue
								.getPushedNotifications();
						for (PushedNotification p : ps) {
							if (p.isSuccessful())
								continue;
							String invalidToken = p.getDevice().getToken();
							System.err.println("unregistering invalid token "
									+ invalidToken);
							unregister(invalidToken);
							Exception theProblem = p.getException();
							theProblem.printStackTrace();
							ResponsePacket theErrorResponse = p.getResponse();
							if (theErrorResponse != null) {
								System.out.println(theErrorResponse
										.getMessage());
							}
						}
						last = new Date().getTime();
						List<Device> inactiveDevices = Push.feedback(
								"push.p12", "pusubi", false);
						for (Device d : inactiveDevices) {
							String invalidToken = d.getToken();
							System.err
									.println("unregistering feedback failed token token "
											+ invalidToken);
							unregister(invalidToken);
						}
					}
					if (job == null)
						continue;
					job.run();
				}
			}
		}
