You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@usergrid.apache.org by sf...@apache.org on 2014/08/18 23:06:21 UTC
[1/8] git commit: add pushy
Repository: incubator-usergrid
Updated Branches:
refs/heads/two-dot-o-push-notifications [created] 2a411bfc7
add pushy
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/d2280dff
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/d2280dff
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/d2280dff
Branch: refs/heads/two-dot-o-push-notifications
Commit: d2280dff201e765e451a5965163125adb449911d
Parents: 3416ac0
Author: Shawn Feldman <sf...@apache.org>
Authored: Mon Aug 18 12:29:44 2014 -0600
Committer: Shawn Feldman <sf...@apache.org>
Committed: Mon Aug 18 12:29:44 2014 -0600
----------------------------------------------------------------------
.../_maven.repositories | 9 +
...14c97c6e3ef40c88590e1b196d3ec55b-javadoc.jar | Bin 0 -> 160121 bytes
...14c97c6e3ef40c88590e1b196d3ec55b-sources.jar | Bin 0 -> 59657 bytes
...63dec68314c97c6e3ef40c88590e1b196d3ec55b.jar | Bin 0 -> 67666 bytes
...7c6e3ef40c88590e1b196d3ec55b.jar.lastUpdated | 16 ++
...63dec68314c97c6e3ef40c88590e1b196d3ec55b.pom | 166 +++++++++++++++++++
...7c6e3ef40c88590e1b196d3ec55b.pom.lastUpdated | 16 ++
.../pushy-0.4-apigee.pom | 166 +++++++++++++++++++
stack/pom.xml | 13 +-
9 files changed, 383 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d2280dff/stack/m2/repository/com/relayrides/pushy/0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b/_maven.repositories
----------------------------------------------------------------------
diff --git a/stack/m2/repository/com/relayrides/pushy/0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b/_maven.repositories b/stack/m2/repository/com/relayrides/pushy/0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b/_maven.repositories
new file mode 100644
index 0000000..a4c5c52
--- /dev/null
+++ b/stack/m2/repository/com/relayrides/pushy/0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b/_maven.repositories
@@ -0,0 +1,9 @@
+#NOTE: This is an internal implementation file, its format can be changed without prior notice.
+#Tue Jul 29 08:10:25 MDT 2014
+pushy-0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b.jar>local-dependencies=
+pushy-0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b.pom>=
+pushy-0.4-apigee.pom>local-dependencies=
+pushy-0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b-sources.jar>=
+pushy-0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b.jar>=
+pushy-0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b.pom>local-dependencies=
+pushy-0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b-javadoc.jar>=
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d2280dff/stack/m2/repository/com/relayrides/pushy/0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b/pushy-0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b-javadoc.jar
----------------------------------------------------------------------
diff --git a/stack/m2/repository/com/relayrides/pushy/0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b/pushy-0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b-javadoc.jar b/stack/m2/repository/com/relayrides/pushy/0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b/pushy-0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b-javadoc.jar
new file mode 100644
index 0000000..f5822ac
Binary files /dev/null and b/stack/m2/repository/com/relayrides/pushy/0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b/pushy-0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b-javadoc.jar differ
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d2280dff/stack/m2/repository/com/relayrides/pushy/0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b/pushy-0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b-sources.jar
----------------------------------------------------------------------
diff --git a/stack/m2/repository/com/relayrides/pushy/0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b/pushy-0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b-sources.jar b/stack/m2/repository/com/relayrides/pushy/0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b/pushy-0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b-sources.jar
new file mode 100644
index 0000000..ea25c9d
Binary files /dev/null and b/stack/m2/repository/com/relayrides/pushy/0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b/pushy-0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b-sources.jar differ
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d2280dff/stack/m2/repository/com/relayrides/pushy/0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b/pushy-0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b.jar
----------------------------------------------------------------------
diff --git a/stack/m2/repository/com/relayrides/pushy/0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b/pushy-0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b.jar b/stack/m2/repository/com/relayrides/pushy/0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b/pushy-0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b.jar
new file mode 100644
index 0000000..1472acb
Binary files /dev/null and b/stack/m2/repository/com/relayrides/pushy/0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b/pushy-0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b.jar differ
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d2280dff/stack/m2/repository/com/relayrides/pushy/0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b/pushy-0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b.jar.lastUpdated
----------------------------------------------------------------------
diff --git a/stack/m2/repository/com/relayrides/pushy/0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b/pushy-0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b.jar.lastUpdated b/stack/m2/repository/com/relayrides/pushy/0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b/pushy-0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b.jar.lastUpdated
new file mode 100644
index 0000000..733d7a6
--- /dev/null
+++ b/stack/m2/repository/com/relayrides/pushy/0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b/pushy-0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b.jar.lastUpdated
@@ -0,0 +1,16 @@
+#NOTE: This is an internal implementation file, its format can be changed without prior notice.
+#Thu Jul 24 13:16:38 MDT 2014
+file\:///Users/apigee/develop/usergrid/usergrid-rest-apigee-production/m2/repository/.error=
+http\://oss.sonatype.org/content/repositories/snapshots/.lastUpdated=1406229398295
+http\://download.java.net/maven/2/.lastUpdated=1406229397326
+http\://repo.maven.apache.org/maven2/.lastUpdated=1406229398556
+http\://oss.sonatype.org/content/repositories/snapshots/.error=
+http\://download.java.net/maven/2/.error=
+file\:///Users/apigee/develop/usergrid/usergrid-rest-apigee-production/m2/repository/.lastUpdated=1406229398434
+http\://repo.maven.apache.org/maven2/.error=
+https\://repository-hector-dev.forge.cloudbees.com/snapshot/.error=
+http\://maven.springframework.org/milestone/.lastUpdated=1406229398432
+http\://repository.codehaus.org/.lastUpdated=1406229397898
+http\://repository.codehaus.org/.error=
+http\://maven.springframework.org/milestone/.error=
+https\://repository-hector-dev.forge.cloudbees.com/snapshot/.lastUpdated=1406229397817
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d2280dff/stack/m2/repository/com/relayrides/pushy/0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b/pushy-0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b.pom
----------------------------------------------------------------------
diff --git a/stack/m2/repository/com/relayrides/pushy/0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b/pushy-0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b.pom b/stack/m2/repository/com/relayrides/pushy/0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b/pushy-0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b.pom
new file mode 100644
index 0000000..982bb21
--- /dev/null
+++ b/stack/m2/repository/com/relayrides/pushy/0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b/pushy-0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b.pom
@@ -0,0 +1,166 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <groupId>com.relayrides</groupId>
+ <artifactId>pushy</artifactId>
+ <version>0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b</version>
+ <name>Pushy</name>
+ <description>A Java library for sending push notifications</description>
+ <parent>
+ <groupId>org.sonatype.oss</groupId>
+ <artifactId>oss-parent</artifactId>
+ <version>7</version>
+ </parent>
+ <licenses>
+ <license>
+ <name>The MIT License (MIT)</name>
+ <url>http://opensource.org/licenses/MIT</url>
+ <distribution>repo</distribution>
+ </license>
+ </licenses>
+ <dependencies>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-api</artifactId>
+ <version>1.7.6</version>
+ </dependency>
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <version>4.11</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>io.netty</groupId>
+ <artifactId>netty-all</artifactId>
+ <version>4.0.21.Final</version>
+ </dependency>
+ <dependency>
+ <groupId>com.googlecode.json-simple</groupId>
+ <artifactId>json-simple</artifactId>
+ <version>1.1.1</version>
+ <exclusions>
+ <exclusion>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-simple</artifactId>
+ <version>1.7.6</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+ <properties>
+ <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+ </properties>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-compiler-plugin</artifactId>
+ <version>3.1</version>
+ <configuration>
+ <source>1.6</source>
+ <target>1.6</target>
+ </configuration>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <version>2.16</version>
+ <configuration>
+ <argLine>-Dorg.slf4j.simpleLogger.defaultLogLevel=warn</argLine>
+ </configuration>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-javadoc-plugin</artifactId>
+ <version>2.9.1</version>
+ <configuration>
+ <overview>${basedir}/src/main/java/overview.html</overview>
+ <show>public</show>
+ <links>
+ <link>http://netty.io/4.0/api/</link>
+ </links>
+ </configuration>
+ <executions>
+ <execution>
+ <id>attach-javadocs</id>
+ <goals>
+ <goal>jar</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-source-plugin</artifactId>
+ <version>2.2.1</version>
+ <executions>
+ <execution>
+ <id>attach-sources</id>
+ <goals>
+ <goal>jar</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+ <profiles>
+ <profile>
+ <id>release-sign-artifacts</id>
+ <activation>
+ <property>
+ <name>performRelease</name>
+ <value>true</value>
+ </property>
+ </activation>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-gpg-plugin</artifactId>
+ <version>1.1</version>
+ <executions>
+ <execution>
+ <id>sign-artifacts</id>
+ <phase>verify</phase>
+ <goals>
+ <goal>sign</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+ </profile>
+ </profiles>
+ <organization>
+ <name>RelayRides</name>
+ <url>https://relayrides.com/</url>
+ </organization>
+ <developers>
+ <developer>
+ <id>jon</id>
+ <name>Jon Chambers</name>
+ <email>jon@relayrides.com</email>
+ <url>https://github.com/jchambers</url>
+ <organization>RelayRides</organization>
+ <organizationUrl>https://relayrides.com/</organizationUrl>
+ <roles>
+ <role>developer</role>
+ </roles>
+ <timezone>-5</timezone>
+ </developer>
+ </developers>
+ <inceptionYear>2013</inceptionYear>
+ <url>http://relayrides.github.com/pushy/</url>
+ <scm>
+ <connection>scm:git:https://github.com/relayrides/pushy.git</connection>
+ <developerConnection>scm:git:git@github.com:relayrides/pushy.git</developerConnection>
+ <url>https://github.com/relayrides/pushy</url>
+ </scm>
+</project>
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d2280dff/stack/m2/repository/com/relayrides/pushy/0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b/pushy-0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b.pom.lastUpdated
----------------------------------------------------------------------
diff --git a/stack/m2/repository/com/relayrides/pushy/0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b/pushy-0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b.pom.lastUpdated b/stack/m2/repository/com/relayrides/pushy/0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b/pushy-0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b.pom.lastUpdated
new file mode 100644
index 0000000..d0ccc7a
--- /dev/null
+++ b/stack/m2/repository/com/relayrides/pushy/0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b/pushy-0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b.pom.lastUpdated
@@ -0,0 +1,16 @@
+#NOTE: This is an internal implementation file, its format can be changed without prior notice.
+#Thu Jul 24 13:16:37 MDT 2014
+file\:///Users/apigee/develop/usergrid/usergrid-rest-apigee-production/m2/repository/.error=
+http\://oss.sonatype.org/content/repositories/snapshots/.lastUpdated=1406229390366
+http\://download.java.net/maven/2/.lastUpdated=1406229388629
+http\://repo.maven.apache.org/maven2/.lastUpdated=1406229397163
+http\://oss.sonatype.org/content/repositories/snapshots/.error=
+http\://download.java.net/maven/2/.error=
+file\:///Users/apigee/develop/usergrid/usergrid-rest-apigee-production/m2/repository/.lastUpdated=1406229396602
+http\://repo.maven.apache.org/maven2/.error=
+https\://repository-hector-dev.forge.cloudbees.com/snapshot/.error=
+http\://maven.springframework.org/milestone/.lastUpdated=1406229396596
+http\://repository.codehaus.org/.lastUpdated=1406229389917
+http\://repository.codehaus.org/.error=
+http\://maven.springframework.org/milestone/.error=
+https\://repository-hector-dev.forge.cloudbees.com/snapshot/.lastUpdated=1406229389776
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d2280dff/stack/m2/repository/com/relayrides/pushy/0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b/pushy-0.4-apigee.pom
----------------------------------------------------------------------
diff --git a/stack/m2/repository/com/relayrides/pushy/0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b/pushy-0.4-apigee.pom b/stack/m2/repository/com/relayrides/pushy/0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b/pushy-0.4-apigee.pom
new file mode 100644
index 0000000..39f61e3
--- /dev/null
+++ b/stack/m2/repository/com/relayrides/pushy/0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b/pushy-0.4-apigee.pom
@@ -0,0 +1,166 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <groupId>com.relayrides</groupId>
+ <artifactId>pushy</artifactId>
+ <version>0.4-apigee</version>
+ <name>Pushy</name>
+ <description>A Java library for sending push notifications</description>
+ <parent>
+ <groupId>org.sonatype.oss</groupId>
+ <artifactId>oss-parent</artifactId>
+ <version>7</version>
+ </parent>
+ <licenses>
+ <license>
+ <name>The MIT License (MIT)</name>
+ <url>http://opensource.org/licenses/MIT</url>
+ <distribution>repo</distribution>
+ </license>
+ </licenses>
+ <dependencies>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-api</artifactId>
+ <version>1.7.6</version>
+ </dependency>
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <version>4.11</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>io.netty</groupId>
+ <artifactId>netty-all</artifactId>
+ <version>4.0.21.Final</version>
+ </dependency>
+ <dependency>
+ <groupId>com.googlecode.json-simple</groupId>
+ <artifactId>json-simple</artifactId>
+ <version>1.1.1</version>
+ <exclusions>
+ <exclusion>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-simple</artifactId>
+ <version>1.7.6</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+ <properties>
+ <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+ </properties>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-compiler-plugin</artifactId>
+ <version>3.1</version>
+ <configuration>
+ <source>1.6</source>
+ <target>1.6</target>
+ </configuration>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <version>2.16</version>
+ <configuration>
+ <argLine>-Dorg.slf4j.simpleLogger.defaultLogLevel=warn</argLine>
+ </configuration>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-javadoc-plugin</artifactId>
+ <version>2.9.1</version>
+ <configuration>
+ <overview>${basedir}/src/main/java/overview.html</overview>
+ <show>public</show>
+ <links>
+ <link>http://netty.io/4.0/api/</link>
+ </links>
+ </configuration>
+ <executions>
+ <execution>
+ <id>attach-javadocs</id>
+ <goals>
+ <goal>jar</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-source-plugin</artifactId>
+ <version>2.2.1</version>
+ <executions>
+ <execution>
+ <id>attach-sources</id>
+ <goals>
+ <goal>jar</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+ <profiles>
+ <profile>
+ <id>release-sign-artifacts</id>
+ <activation>
+ <property>
+ <name>performRelease</name>
+ <value>true</value>
+ </property>
+ </activation>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-gpg-plugin</artifactId>
+ <version>1.1</version>
+ <executions>
+ <execution>
+ <id>sign-artifacts</id>
+ <phase>verify</phase>
+ <goals>
+ <goal>sign</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+ </profile>
+ </profiles>
+ <organization>
+ <name>RelayRides</name>
+ <url>https://relayrides.com/</url>
+ </organization>
+ <developers>
+ <developer>
+ <id>jon</id>
+ <name>Jon Chambers</name>
+ <email>jon@relayrides.com</email>
+ <url>https://github.com/jchambers</url>
+ <organization>RelayRides</organization>
+ <organizationUrl>https://relayrides.com/</organizationUrl>
+ <roles>
+ <role>developer</role>
+ </roles>
+ <timezone>-5</timezone>
+ </developer>
+ </developers>
+ <inceptionYear>2013</inceptionYear>
+ <url>http://relayrides.github.com/pushy/</url>
+ <scm>
+ <connection>scm:git:https://github.com/relayrides/pushy.git</connection>
+ <developerConnection>scm:git:git@github.com:relayrides/pushy.git</developerConnection>
+ <url>https://github.com/relayrides/pushy</url>
+ </scm>
+</project>
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d2280dff/stack/pom.xml
----------------------------------------------------------------------
diff --git a/stack/pom.xml b/stack/pom.xml
index e51da28..35c3979 100644
--- a/stack/pom.xml
+++ b/stack/pom.xml
@@ -1400,13 +1400,20 @@
<version>1.1</version>
</dependency>
- <dependency>
+ <dependency>
<groupId>org.elasticsearch</groupId>
<artifactId>elasticsearch</artifactId>
<version>${elasticsearch.version}</version>
- </dependency>
+ </dependency>
+
+ <dependency>
+ <groupId>com.relayrides</groupId>
+ <artifactId>pushy</artifactId>
+ <!-- The sha in the version is the git commit used in this build. Check out the pushy source, then this commit to build the library locally -->
+ <version>0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b</version>
+ </dependency>
- </dependencies>
+ </dependencies>
</dependencyManagement>
[3/8] git commit: add notification entities
Posted by sf...@apache.org.
add notification entities
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/05d90c66
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/05d90c66
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/05d90c66
Branch: refs/heads/two-dot-o-push-notifications
Commit: 05d90c6680363dedd1d3aae0c7587bd7b6b4a55f
Parents: 9d2121c
Author: Shawn Feldman <sf...@apache.org>
Authored: Mon Aug 18 13:49:29 2014 -0600
Committer: Shawn Feldman <sf...@apache.org>
Committed: Mon Aug 18 13:49:29 2014 -0600
----------------------------------------------------------------------
.../usergrid/persistence/Notification.java | 241 +++++++++++++++++++
.../apache/usergrid/persistence/Notifier.java | 139 +++++++++++
.../apache/usergrid/persistence/Receipt.java | 133 ++++++++++
3 files changed, 513 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/05d90c66/stack/core/src/main/java/org/apache/usergrid/persistence/Notification.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/persistence/Notification.java b/stack/core/src/main/java/org/apache/usergrid/persistence/Notification.java
new file mode 100644
index 0000000..f3b7392
--- /dev/null
+++ b/stack/core/src/main/java/org/apache/usergrid/persistence/Notification.java
@@ -0,0 +1,241 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.usergrid.persistence;
+
+import org.codehaus.jackson.annotate.JsonIgnore;
+import org.codehaus.jackson.map.annotate.JsonSerialize;
+import org.codehaus.jackson.map.annotate.JsonSerialize.Inclusion;
+
+import javax.xml.bind.annotation.XmlRootElement;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+import org.apache.usergrid.persistence.PathQuery;
+import org.apache.usergrid.persistence.TypedEntity;
+import org.apache.usergrid.persistence.annotations.EntityCollection;
+import org.apache.usergrid.persistence.annotations.EntityProperty;
+import org.apache.usergrid.persistence.entities.Device;
+
+/**
+ * The entity class for representing Notifications.
+ */
+@XmlRootElement
+public class Notification extends TypedEntity {
+
+ public static final String ENTITY_TYPE = "notification";
+
+ public static final String RECEIPTS_COLLECTION = "receipts";
+
+ public static enum State {
+ CREATED, FAILED, SCHEDULED, STARTED, FINISHED, CANCELED, EXPIRED
+ }
+
+ /** Map Notifier ID -> Payload data */
+ @EntityProperty
+ protected Map<String, Object> payloads;
+
+ /** Time processed */
+ @EntityProperty
+ protected Long queued;
+
+ /** Time send started */
+ @EntityProperty
+ protected Long started;
+
+ /** Time processed */
+ @EntityProperty
+ protected Long finished;
+
+ /** Time to deliver to provider */
+ @EntityProperty
+ protected Long deliver;
+
+ /** Time to expire the notification */
+ @EntityProperty
+ protected Long expire;
+
+ /** True if notification is canceled */
+ @EntityProperty
+ protected Boolean canceled;
+
+ /** Error message */
+ @EntityProperty
+ protected String errorMessage;
+
+ @EntityCollection(type = "receipt")
+ protected List<UUID> receipts;
+
+ /** stats (sent & errors) */
+ @EntityProperty
+ protected Map<String, Long> statistics;
+
+ /** stats (sent & errors) */
+ @EntityProperty
+ @JsonIgnore
+ protected PathQuery<Device> pathQuery;
+
+ public Notification() {
+ }
+
+ @JsonIgnore
+ public List<UUID> getReceipts() {
+ return receipts;
+ }
+
+ public void setReceipts(List<UUID> receipts) {
+ this.receipts = receipts;
+ }
+
+ @JsonSerialize(include = Inclusion.NON_NULL)
+ public Map<String, Object> getPayloads() {
+ return payloads;
+ }
+
+ public void setPayloads(Map<String, Object> payloads) {
+ this.payloads = payloads;
+ }
+
+ @JsonSerialize(include = Inclusion.NON_NULL)
+ public Long getFinished() {
+ return finished;
+ }
+
+ public void setFinished(Long finished) {
+ this.finished = finished;
+ }
+
+ @JsonSerialize(include = Inclusion.NON_NULL)
+ public Long getDeliver() {
+ return deliver;
+ }
+
+ public void setDeliver(Long deliver) {
+ this.deliver = deliver;
+ }
+
+ @JsonSerialize(include = Inclusion.NON_NULL)
+ public Long getExpire() {
+ return expire;
+ }
+
+ public void setExpire(Long expire) {
+ this.expire = expire;
+ }
+
+ @JsonIgnore
+ public boolean isExpired() {
+ return expire != null && expire > System.currentTimeMillis();
+ }
+
+ @JsonSerialize(include = Inclusion.NON_NULL)
+ public Boolean getCanceled() {
+ return canceled;
+ }
+
+ public void setCanceled(Boolean canceled) {
+ this.canceled = canceled;
+ }
+
+ @JsonSerialize(include = Inclusion.NON_NULL)
+ public Long getStarted() {
+ return started;
+ }
+
+ public void setStarted(Long started) {
+ this.started = started;
+ }
+
+ @JsonSerialize(include = Inclusion.NON_NULL)
+ public String getErrorMessage() {
+ return errorMessage;
+ }
+
+ public void setErrorMessage(String errorMessage) {
+ this.errorMessage = errorMessage;
+ }
+
+ @JsonSerialize(include = Inclusion.NON_NULL)
+ public Map<String, Long> getStatistics() {
+ return statistics;
+ }
+
+ public void setStatistics(Map<String, Long> statistics) {
+ this.statistics = statistics;
+ }
+
+ public void updateStatistics(long sent, long errors) {
+ if (this.statistics == null) {
+ this.statistics = new HashMap<String, Long>(2);
+ this.statistics.put("sent", sent);
+ this.statistics.put("errors", errors);
+ } else {
+ this.statistics.put("sent", sent + this.statistics.get("sent"));
+ this.statistics.put("errors",
+ errors + this.statistics.get("errors"));
+ }
+ }
+
+ /** don't bother, I will ignore you */
+ public void setState(State ignored) {
+ // does nothing - state is derived
+ }
+
+ @EntityProperty
+ public State getState() {
+ if (getErrorMessage() != null) {
+ return State.FAILED;
+ } else if (getCanceled() == Boolean.TRUE) {
+ return State.CANCELED;
+ } else if (getFinished() != null) {
+ return State.FINISHED;
+ } else if (getStarted() != null && getDeliver() == null) {
+ return State.STARTED;
+ } else if (isExpired()) {
+ return State.EXPIRED;
+ } else if (getDeliver() != null || getQueued() != null) {
+ return State.SCHEDULED;
+ }
+ return State.CREATED;
+ }
+
+ @JsonIgnore
+ public PathQuery<Device> getPathQuery() {
+ return pathQuery;
+ }
+
+ public void setPathQuery(PathQuery<Device> pathQuery) {
+ this.pathQuery = pathQuery;
+ }
+
+ @JsonIgnore
+ public int getExpireTimeInSeconds() {
+ long expirySeconds = getExpire() != null ? getExpire() * 1000 : 0;
+ return (expirySeconds > Integer.MAX_VALUE) ? Integer.MAX_VALUE
+ : (int) expirySeconds;
+ }
+
+ @JsonSerialize(include = Inclusion.NON_NULL)
+ public Long getQueued() {
+ return queued;
+ }
+
+ public void setQueued(Long queued) {
+ this.queued = queued;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/05d90c66/stack/core/src/main/java/org/apache/usergrid/persistence/Notifier.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/persistence/Notifier.java b/stack/core/src/main/java/org/apache/usergrid/persistence/Notifier.java
new file mode 100644
index 0000000..953ec6d
--- /dev/null
+++ b/stack/core/src/main/java/org/apache/usergrid/persistence/Notifier.java
@@ -0,0 +1,139 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.usergrid.persistence;
+
+import org.codehaus.jackson.annotate.JsonIgnore;
+import org.codehaus.jackson.map.annotate.JsonSerialize;
+import org.codehaus.jackson.map.annotate.JsonSerialize.Inclusion;
+import org.apache.usergrid.persistence.TypedEntity;
+import org.apache.usergrid.persistence.annotations.EntityProperty;
+
+import javax.net.ssl.SSLContext;
+import javax.xml.bind.annotation.XmlRootElement;
+import java.io.ByteArrayInputStream;
+import java.io.InputStream;
+import java.security.KeyStore;
+import java.util.Random;
+import java.util.UUID;
+
+/**
+ * The entity class for representing Notifiers.
+ */
+@XmlRootElement
+public class Notifier extends TypedEntity {
+
+ public static final String ENTITY_TYPE = "notifier";
+
+ @EntityProperty(aliasProperty = true, unique = true, basic = true)
+ protected String name;
+
+ @EntityProperty(required = true)
+ protected String provider;
+
+ @EntityProperty
+ protected String environment;
+
+ // Apple APNs
+ @EntityProperty(indexed = false, includedInExport = false, encrypted = true)
+ protected byte[] p12Certificate;
+
+ // Apple APNs
+ @EntityProperty(indexed = false, includedInExport = false, encrypted = true)
+ protected String certificatePassword;
+
+ // Google GCM
+ @EntityProperty(indexed = false, includedInExport = false, encrypted = true)
+ protected String apiKey;
+ private javax.net.ssl.SSLContext SSLContext;
+
+ public Notifier() {
+ }
+
+ public Notifier(UUID id) {
+ uuid = id;
+ }
+
+
+
+ @Override
+ @JsonSerialize(include = Inclusion.NON_NULL)
+ public String getName() {
+ return name;
+ }
+
+ public void setName(String name) {
+ this.name = name;
+ }
+
+ @JsonSerialize(include = Inclusion.NON_NULL)
+ public String getProvider() {
+ return provider;
+ }
+
+ public void setProvider(String provider) {
+ this.provider = provider;
+ }
+
+ @JsonSerialize(include = Inclusion.NON_NULL)
+ public String getEnvironment() {
+ return environment;
+ }
+
+ public void setEnvironment(String environment) {
+ this.environment = environment;
+ }
+
+ @JsonIgnore
+ public boolean isProduction() {
+ return !"development".equals(environment);
+ }
+
+ @JsonIgnore
+ public byte[] getP12Certificate() {
+ return p12Certificate;
+ }
+
+ public void setP12Certificate(byte[] p12Certificate) {
+ this.p12Certificate = p12Certificate;
+ }
+
+ @JsonIgnore
+ public InputStream getP12CertificateStream() {
+ byte[] cert = getP12Certificate();
+ return cert != null ? new ByteArrayInputStream(cert) : null;
+ }
+
+ @JsonIgnore
+ public String getCertificatePassword() {
+ return certificatePassword;
+ }
+
+ public void setCertificatePassword(String certificatePassword) {
+ this.certificatePassword = certificatePassword;
+ }
+
+ @JsonIgnore
+ public String getApiKey() {
+ return apiKey;
+ }
+
+ public void setApiKey(String apiKey) {
+ this.apiKey = apiKey;
+ }
+
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/05d90c66/stack/core/src/main/java/org/apache/usergrid/persistence/Receipt.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/persistence/Receipt.java b/stack/core/src/main/java/org/apache/usergrid/persistence/Receipt.java
new file mode 100644
index 0000000..f1c034b
--- /dev/null
+++ b/stack/core/src/main/java/org/apache/usergrid/persistence/Receipt.java
@@ -0,0 +1,133 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.usergrid.persistence;
+
+import org.codehaus.jackson.map.annotate.JsonSerialize;
+import org.codehaus.jackson.map.annotate.JsonSerialize.Inclusion;
+
+import javax.xml.bind.annotation.XmlRootElement;
+import java.util.UUID;
+import org.apache.usergrid.persistence.annotations.EntityProperty;
+
+@XmlRootElement
+public class Receipt extends TypedEntity {
+
+ public static final String ENTITY_TYPE = "receipt";
+ public static final String NOTIFICATION_CONNECTION = "notification";
+
+ /** device id **/
+ @EntityProperty
+ protected UUID deviceId;
+
+ /** data sent to provider */
+ @EntityProperty
+ protected Object payload;
+
+ /** Time sent to provider */
+ @EntityProperty
+ protected Long sent;
+
+ /** Error code */
+ @EntityProperty
+ protected Object errorCode;
+
+ /** Error message */
+ @EntityProperty
+ protected String errorMessage;
+
+ /** The push token given by the provider */
+ @EntityProperty
+ protected String notifierId;
+
+ /**
+ * UUID of the Notification that sent this - not a Connection for
+ * performance reasons
+ */
+ @EntityProperty
+ protected UUID notificationUUID;
+
+ public Receipt() {
+ }
+
+ public Receipt(UUID notificationUUID, String notifierId, Object payload,UUID deviceId) {
+ this.notificationUUID = notificationUUID;
+ this.notifierId = notifierId;
+ this.payload = payload;
+ this.setDeviceId(deviceId);
+ }
+
+ @JsonSerialize(include = Inclusion.NON_NULL)
+ public Object getPayload() {
+ return payload;
+ }
+
+ public void setPayload(Object payload) {
+ this.payload = payload;
+ }
+
+ @JsonSerialize(include = Inclusion.NON_NULL)
+ public Long getSent() {
+ return sent;
+ }
+
+ public void setSent(Long sent) {
+ this.sent = sent;
+ }
+
+ @JsonSerialize(include = Inclusion.NON_NULL)
+ public Object getErrorCode() {
+ return errorCode;
+ }
+
+ public void setErrorCode(Object errorCode) {
+ this.errorCode = errorCode;
+ }
+
+ @JsonSerialize(include = Inclusion.NON_NULL)
+ public String getErrorMessage() {
+ return errorMessage;
+ }
+
+ public void setErrorMessage(String errorMessage) {
+ this.errorMessage = errorMessage;
+ }
+
+ public String getNotifierId() {
+ return notifierId;
+ }
+
+ public void setNotifierId(String notifierId) {
+ this.notifierId = notifierId;
+ }
+
+ public UUID getNotificationUUID() {
+ return notificationUUID;
+ }
+
+ public void setNotificationUUID(UUID notificationUUID) {
+ this.notificationUUID = notificationUUID;
+ }
+
+ @JsonSerialize(include = Inclusion.NON_NULL)
+ public UUID getDeviceId() {
+ return deviceId;
+ }
+
+ public void setDeviceId(UUID deviceId) {
+ this.deviceId = deviceId;
+ }
+}
[5/8] git commit: adding an iterator interface to look for extra pages
Posted by sf...@apache.org.
adding an iterator interface to look for extra pages
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/0a61101d
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/0a61101d
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/0a61101d
Branch: refs/heads/two-dot-o-push-notifications
Commit: 0a61101dafc1ab5e719fe3a8662714ab32b89b19
Parents: 99c80fa
Author: Shawn Feldman <sf...@apache.org>
Authored: Tue Jul 29 11:30:55 2014 -0600
Committer: Shawn Feldman <sf...@apache.org>
Committed: Mon Aug 18 14:37:19 2014 -0600
----------------------------------------------------------------------
.../persistence/MultiQueryIterator.java | 6 +++-
.../persistence/PagingResultsIterator.java | 7 ++++-
.../usergrid/persistence/ResultsIterator.java | 30 ++++++++++++++++++++
3 files changed, 41 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/0a61101d/stack/core/src/main/java/org/apache/usergrid/persistence/MultiQueryIterator.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/persistence/MultiQueryIterator.java b/stack/core/src/main/java/org/apache/usergrid/persistence/MultiQueryIterator.java
index 6135524..5b64d0b 100644
--- a/stack/core/src/main/java/org/apache/usergrid/persistence/MultiQueryIterator.java
+++ b/stack/core/src/main/java/org/apache/usergrid/persistence/MultiQueryIterator.java
@@ -27,7 +27,7 @@ import org.apache.usergrid.persistence.index.query.Query.Level;
* For each in a set of source refs executes a sub-query and provides a unified iterator over
* the union of all results. Honors page sizes for the Query to ensure memory isn't blown out.
*/
-public class MultiQueryIterator implements Iterator {
+public class MultiQueryIterator implements ResultsIterator {
private EntityManager entityManager;
private Iterator<EntityRef> source;
@@ -74,6 +74,10 @@ public class MultiQueryIterator implements Iterator {
return false;
}
+ @Override
+ public boolean hasPages(){
+ return currentIterator != null && currentIterator instanceof ResultsIterator && ((ResultsIterator)currentIterator).hasPages();
+ }
@Override
public Object next() {
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/0a61101d/stack/core/src/main/java/org/apache/usergrid/persistence/PagingResultsIterator.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/persistence/PagingResultsIterator.java b/stack/core/src/main/java/org/apache/usergrid/persistence/PagingResultsIterator.java
index eab6c37..183cc30 100644
--- a/stack/core/src/main/java/org/apache/usergrid/persistence/PagingResultsIterator.java
+++ b/stack/core/src/main/java/org/apache/usergrid/persistence/PagingResultsIterator.java
@@ -25,7 +25,7 @@ import static org.apache.usergrid.persistence.index.query.Query.Level.REFS;
/** iterates over a Results object, crossing page boundaries automatically */
-public class PagingResultsIterator implements Iterator, Iterable {
+public class PagingResultsIterator implements ResultsIterator, Iterable {
private Results results;
private Iterator currentPageIterator;
@@ -61,6 +61,11 @@ public class PagingResultsIterator implements Iterator, Iterable {
return false;
}
+ @Override
+ public boolean hasPages(){
+ return results != null && results.hasCursor();
+ }
+
/** @return the next object (type varies according the Results.Level) */
@Override
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/0a61101d/stack/core/src/main/java/org/apache/usergrid/persistence/ResultsIterator.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/persistence/ResultsIterator.java b/stack/core/src/main/java/org/apache/usergrid/persistence/ResultsIterator.java
new file mode 100644
index 0000000..6ed2bad
--- /dev/null
+++ b/stack/core/src/main/java/org/apache/usergrid/persistence/ResultsIterator.java
@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.usergrid.persistence;
+
+import java.util.Iterator;
+
+/**
+ * Iterator to know if we have more records in cursor.
+ */
+public interface ResultsIterator extends Iterator {
+ /**
+ * is there a cursor
+ * @return
+ */
+ boolean hasPages();
+}
[8/8] git commit: adding rest resources
Posted by sf...@apache.org.
adding rest resources
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/2a411bfc
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/2a411bfc
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/2a411bfc
Branch: refs/heads/two-dot-o-push-notifications
Commit: 2a411bfc78391c470cfe7ceb84488aa15429ead4
Parents: 9d7901a
Author: Shawn Feldman <sf...@apache.org>
Authored: Mon Aug 18 15:06:06 2014 -0600
Committer: Shawn Feldman <sf...@apache.org>
Committed: Mon Aug 18 15:06:06 2014 -0600
----------------------------------------------------------------------
.../notifiers/NotifierResource.java | 103 +++++++++++++
.../notifiers/NotifiersResource.java | 153 +++++++++++++++++++
2 files changed, 256 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2a411bfc/stack/rest/src/main/java/org/apache/usergrid/rest/applications/notifiers/NotifierResource.java
----------------------------------------------------------------------
diff --git a/stack/rest/src/main/java/org/apache/usergrid/rest/applications/notifiers/NotifierResource.java b/stack/rest/src/main/java/org/apache/usergrid/rest/applications/notifiers/NotifierResource.java
new file mode 100644
index 0000000..72883b0
--- /dev/null
+++ b/stack/rest/src/main/java/org/apache/usergrid/rest/applications/notifiers/NotifierResource.java
@@ -0,0 +1,103 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.usergrid.rest.applications.notifiers;
+
+import com.sun.jersey.api.json.JSONWithPadding;
+import com.sun.jersey.multipart.FormDataMultiPart;
+import java.io.InputStream;
+import org.apache.commons.io.IOUtils;
+import org.apache.usergrid.persistence.index.query.Identifier;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.context.annotation.Scope;
+import org.springframework.stereotype.Component;
+
+import javax.ws.rs.*;
+import javax.ws.rs.core.Context;
+import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.UriInfo;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+
+import org.apache.usergrid.rest.ApiResponse;
+import org.apache.usergrid.rest.applications.ServiceResource;
+import org.apache.usergrid.rest.security.annotations.RequireApplicationAccess;
+import org.apache.usergrid.services.ServiceAction;
+import org.apache.usergrid.services.ServicePayload;
+
+@Component("org.apache.usergrid.rest.applications.notifiers.NotifierResource")
+@Scope("prototype")
+@Produces(MediaType.APPLICATION_JSON)
+public class NotifierResource extends ServiceResource {
+
+ private static final Logger logger = LoggerFactory.getLogger(NotifierResource.class);
+
+ private Identifier identifier;
+
+ public NotifierResource init(Identifier identifier) throws Exception {
+ this.identifier = identifier;
+ return this;
+ }
+
+ /* Multipart PUT update with uploaded p12Certificate */
+ @PUT
+ @RequireApplicationAccess
+ @Consumes(MediaType.MULTIPART_FORM_DATA)
+ @Override
+ public JSONWithPadding executeMultiPartPut(@Context UriInfo ui,
+ @QueryParam("callback") @DefaultValue("callback") String callback,
+ FormDataMultiPart multiPart) throws Exception {
+
+ logger.debug("NotifierResource.executePut");
+
+ String name = getValueOrNull(multiPart, "name");
+ String provider = getValueOrNull(multiPart, "provider");
+ String certPassword = getValueOrNull(multiPart, "certificatePassword");
+
+ InputStream is = null;
+ if (multiPart.getField("p12Certificate") != null) {
+ is = multiPart.getField("p12Certificate").getEntityAs(InputStream.class);
+ }
+
+ HashMap<String, Object> properties = new LinkedHashMap<String, Object>();
+ properties.put("name", name);
+ properties.put("provider", provider);
+ properties.put("environment", "production");
+ properties.put("certificatePassword", certPassword);
+ if (is != null) {
+ byte[] certBytes = IOUtils.toByteArray(is);
+ properties.put("p12Certificate", certBytes);
+ }
+
+ ApiResponse response = createApiResponse();
+ response.setAction("put");
+ response.setApplication(services.getApplication());
+ response.setParams(ui.getQueryParameters());
+ ServicePayload payload = getPayload(properties);
+ executeServiceRequest(ui, response, ServiceAction.PUT, payload);
+
+ return new JSONWithPadding(response, callback);
+ }
+
+ private String getValueOrNull(FormDataMultiPart multiPart, String name) {
+ if (multiPart.getField(name) != null) {
+ return multiPart.getField(name).getValue();
+ }
+ return null;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/2a411bfc/stack/rest/src/main/java/org/apache/usergrid/rest/applications/notifiers/NotifiersResource.java
----------------------------------------------------------------------
diff --git a/stack/rest/src/main/java/org/apache/usergrid/rest/applications/notifiers/NotifiersResource.java b/stack/rest/src/main/java/org/apache/usergrid/rest/applications/notifiers/NotifiersResource.java
new file mode 100644
index 0000000..cfbbb45
--- /dev/null
+++ b/stack/rest/src/main/java/org/apache/usergrid/rest/applications/notifiers/NotifiersResource.java
@@ -0,0 +1,153 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.usergrid.rest.applications.notifiers;
+
+import com.sun.jersey.api.json.JSONWithPadding;
+import com.sun.jersey.multipart.FormDataMultiPart;
+import java.io.InputStream;
+import java.util.HashMap;
+import org.apache.commons.io.IOUtils;
+import org.apache.usergrid.persistence.index.query.Identifier;
+import org.apache.usergrid.persistence.index.query.Query;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.context.annotation.Scope;
+import org.springframework.stereotype.Component;
+
+
+import javax.ws.rs.*;
+import javax.ws.rs.core.Context;
+import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.PathSegment;
+import javax.ws.rs.core.UriInfo;
+import java.util.LinkedHashMap;
+import java.util.UUID;
+
+import org.apache.usergrid.rest.AbstractContextResource;
+import org.apache.usergrid.rest.ApiResponse;
+import org.apache.usergrid.rest.applications.ServiceResource;
+import org.apache.usergrid.rest.security.annotations.RequireApplicationAccess;
+import org.apache.usergrid.services.ServiceAction;
+import org.apache.usergrid.services.ServicePayload;
+
+import static org.apache.usergrid.services.ServiceParameter.addParameter;
+
+@Component("org.apache.usergrid.rest.applications.notifiers.NotifiersResource")
+@Scope("prototype")
+@Produces(MediaType.APPLICATION_JSON)
+public class NotifiersResource extends ServiceResource {
+
+ private static final Logger logger = LoggerFactory
+ .getLogger(NotifiersResource.class);
+
+ @Override
+ @Path("{entityId: [A-Fa-f0-9]{8}-[A-Fa-f0-9]{4}-[A-Fa-f0-9]{4}-[A-Fa-f0-9]{4}-[A-Fa-f0-9]{12}}")
+ public AbstractContextResource addIdParameter(@Context UriInfo ui,
+ @PathParam("entityId") PathSegment entityId) throws Exception {
+
+ logger.info("NotifiersResource.addIdParameter");
+
+ UUID itemId = UUID.fromString(entityId.getPath());
+
+ addParameter(getServiceParameters(), itemId);
+
+ addMatrixParams(getServiceParameters(), ui, entityId);
+
+ return getSubResource(NotifierResource.class).init( Identifier.fromUUID(itemId));
+ }
+
+ @Override
+ @Path("{itemName}")
+ public AbstractContextResource addNameParameter(@Context UriInfo ui,
+ @PathParam("itemName") PathSegment itemName) throws Exception {
+
+ logger.info("NotifiersResource.addNameParameter");
+
+ logger.info("Current segment is " + itemName.getPath());
+
+ if (itemName.getPath().startsWith("{")) {
+ Query query = Query.fromJsonString(itemName.getPath());
+ if (query != null) {
+ addParameter(getServiceParameters(), query);
+ }
+ addMatrixParams(getServiceParameters(), ui, itemName);
+
+ return getSubResource(ServiceResource.class);
+ }
+
+ addParameter(getServiceParameters(), itemName.getPath());
+
+ addMatrixParams(getServiceParameters(), ui, itemName);
+ Identifier id = Identifier.from(itemName.getPath());
+ if (id == null) {
+ throw new IllegalArgumentException(
+ "Not a valid Notifier identifier: " + itemName.getPath());
+ }
+ return getSubResource(NotifierResource.class).init(id);
+ }
+
+ /* Multipart POST create with uploaded p12Certificate */
+ @POST
+ @RequireApplicationAccess
+ @Consumes(MediaType.MULTIPART_FORM_DATA)
+ @Override
+ public JSONWithPadding executeMultiPartPost(
+ @Context UriInfo ui,
+ @QueryParam("callback") @DefaultValue("callback") String callback,
+ FormDataMultiPart multiPart)
+ throws Exception {
+
+ logger.debug("ServiceResource.uploadData");
+
+ String name = getValueOrNull(multiPart, "name");
+ String provider = getValueOrNull(multiPart, "provider");
+ String environment = getValueOrNull(multiPart, "environment");
+ String certPassword = getValueOrNull(multiPart, "certificatePassword");
+
+ InputStream is = null;
+ if (multiPart.getField("p12Certificate") != null) {
+ is = multiPart.getField("p12Certificate").getEntityAs(InputStream.class);
+ }
+
+
+ HashMap<String, Object> certProps = new LinkedHashMap<String, Object>();
+ certProps.put("name", name);
+ certProps.put("provider", provider);
+ certProps.put("environment", environment);
+ certProps.put("certificatePassword", certPassword);
+ if (is != null) {
+ byte[] certBytes = IOUtils.toByteArray(is);
+ certProps.put("p12Certificate", certBytes);
+ }
+
+ ApiResponse response = createApiResponse();
+ response.setAction("post");
+ response.setApplication(services.getApplication());
+ response.setParams(ui.getQueryParameters());
+ ServicePayload payload = getPayload(certProps);
+ executeServiceRequest(ui, response, ServiceAction.POST, payload);
+
+ return new JSONWithPadding(response, callback);
+ }
+
+ private String getValueOrNull(FormDataMultiPart multiPart, String name) {
+ if (multiPart.getField(name) != null) {
+ return multiPart.getField(name).getValue();
+ }
+ return null;
+ }
+}
[4/8] git commit: add metrics factory, cod hale.metrics to pom,
add property
Posted by sf...@apache.org.
add metrics factory, cod hale.metrics to pom, add property
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/99c80faf
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/99c80faf
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/99c80faf
Branch: refs/heads/two-dot-o-push-notifications
Commit: 99c80faf21e179ab83cf9d1c8090a52ab3b9c481
Parents: 05d90c6
Author: Shawn Feldman <sf...@apache.org>
Authored: Fri Jul 18 14:41:58 2014 -0600
Committer: Shawn Feldman <sf...@apache.org>
Committed: Mon Aug 18 14:10:44 2014 -0600
----------------------------------------------------------------------
.../main/resources/usergrid-default.properties | 3 +
stack/core/pom.xml | 11 +-
.../apache/usergrid/metrics/MetricsFactory.java | 121 +++++++++++++++++++
.../main/resources/usergrid-core-context.xml | 3 +
stack/pom.xml | 2 +-
5 files changed, 138 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/99c80faf/stack/config/src/main/resources/usergrid-default.properties
----------------------------------------------------------------------
diff --git a/stack/config/src/main/resources/usergrid-default.properties b/stack/config/src/main/resources/usergrid-default.properties
index 09d4f94..1f7caf5 100644
--- a/stack/config/src/main/resources/usergrid-default.properties
+++ b/stack/config/src/main/resources/usergrid-default.properties
@@ -365,6 +365,9 @@ usergrid.management.email.user-pin=\
<p>${pin}</p>
+# graphite server
+usergrid.metrics.graphite.host=badhost
+
###############################################################################
#
# Redirect urls to use instead of internal JSPs. Not all of these should be
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/99c80faf/stack/core/pom.xml
----------------------------------------------------------------------
diff --git a/stack/core/pom.xml b/stack/core/pom.xml
index b3a3949..ac94015 100644
--- a/stack/core/pom.xml
+++ b/stack/core/pom.xml
@@ -489,7 +489,6 @@
<scope>test</scope>
</dependency>
-
<!-- Core Persistence deps -->
<dependency>
<groupId>org.apache.usergrid</groupId>
@@ -548,6 +547,16 @@
<version>2.0.0-SNAPSHOT</version>
</dependency>-->
+ <dependency>
+ <groupId>com.codahale.metrics</groupId>
+ <artifactId>metrics-core</artifactId>
+ <version>${metrics.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>com.codahale.metrics</groupId>
+ <artifactId>metrics-graphite</artifactId>
+ <version>${metrics.version}</version>
+ </dependency>
</dependencies>
</project>
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/99c80faf/stack/core/src/main/java/org/apache/usergrid/metrics/MetricsFactory.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/metrics/MetricsFactory.java b/stack/core/src/main/java/org/apache/usergrid/metrics/MetricsFactory.java
new file mode 100644
index 0000000..c5dbf1e
--- /dev/null
+++ b/stack/core/src/main/java/org/apache/usergrid/metrics/MetricsFactory.java
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.usergrid.metrics;
+import com.codahale.metrics.*;
+import com.codahale.metrics.graphite.Graphite;
+import com.codahale.metrics.graphite.GraphiteReporter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.PostConstruct;
+import java.net.InetSocketAddress;
+import java.util.Properties;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Singleton class to manage metrics.
+ */
+@Component("metricsFactory")
+public class MetricsFactory {
+ @Autowired
+ private Properties properties;
+ public MetricRegistry registry;
+ private GraphiteReporter graphiteReporter;
+ private JmxReporter jmxReporter;
+ private ConcurrentHashMap<String,Metric> hashMap;
+ private static final Logger LOG = LoggerFactory.getLogger(MetricsFactory.class);
+
+ public MetricsFactory() {
+
+ }
+
+ @PostConstruct
+ void init() {
+ properties = new Properties();
+ try {
+ properties.load(Thread.currentThread()
+ .getContextClassLoader()
+ .getResourceAsStream("usergrid.properties"));
+ } catch (Exception e) {
+ LOG.error("Could not load props","");
+ }
+ registry = new MetricRegistry();
+ String badHost = "badhost";
+ String metricsHost = properties.getProperty("usergrid.metrics.graphite.host", badHost);
+ Graphite graphite = new Graphite(new InetSocketAddress(metricsHost, 2003));
+ graphiteReporter = GraphiteReporter.forRegistry(registry)
+ .prefixedWith("notifications")
+ .convertRatesTo(TimeUnit.SECONDS)
+ .convertDurationsTo(TimeUnit.MILLISECONDS)
+ .filter(MetricFilter.ALL)
+ .build(graphite);
+ if(metricsHost!=badHost) {
+ graphiteReporter.start(30, TimeUnit.SECONDS);
+ }else {
+ LOG.warn("MetricsService:Logger not started.");
+ graphiteReporter.stop();
+ }
+ hashMap = new ConcurrentHashMap<String, Metric>();
+
+ jmxReporter = JmxReporter.forRegistry(registry).build();
+ jmxReporter.start();
+ }
+
+ public MetricRegistry getRegistry() {
+ return registry;
+ }
+
+ public Timer getTimer(Class<?> klass, String name) {
+ return getMetric(Timer.class, klass, name);
+ }
+
+ public Histogram getHistogram(Class<?> klass, String name) {
+ return getMetric(Histogram.class, klass, name);
+ }
+
+ public Counter getCounter(Class<?> klass, String name) {
+ return getMetric(Counter.class, klass, name);
+ }
+
+ public Meter getMeter(Class<?> klass, String name) {
+ return getMetric(Meter.class, klass, name);
+ }
+
+ private <T> T getMetric(Class<T> metricClass, Class<?> klass, String name) {
+ String key = metricClass.getName() + klass.getName() + name;
+ Metric metric = hashMap.get(key);
+ if (metric == null) {
+ if (metricClass == Histogram.class) {
+ metric = this.getRegistry().histogram(MetricRegistry.name(klass, name));
+ }
+ if (metricClass == Timer.class) {
+ metric = this.getRegistry().timer(MetricRegistry.name(klass, name));
+ }
+ if (metricClass == Meter.class) {
+ metric = this.getRegistry().meter(MetricRegistry.name(klass, name));
+ }
+ if (metricClass == Counter.class) {
+ metric = this.getRegistry().counter(MetricRegistry.name(klass, name));
+ }
+ hashMap.put(key, metric);
+ }
+ return (T) metric;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/99c80faf/stack/core/src/main/resources/usergrid-core-context.xml
----------------------------------------------------------------------
diff --git a/stack/core/src/main/resources/usergrid-core-context.xml b/stack/core/src/main/resources/usergrid-core-context.xml
index 9321fb6..cd8802c 100644
--- a/stack/core/src/main/resources/usergrid-core-context.xml
+++ b/stack/core/src/main/resources/usergrid-core-context.xml
@@ -186,6 +186,9 @@
<bean id="jobFactory" class="org.apache.usergrid.batch.UsergridJobFactory" />
+ <bean id="metricsFactory" class="org.apache.usergrid.metrics.MetricsFactory" scope="singleton"/>
+
+ <!-- scan all job classes -->
<context:component-scan base-package="org.apache.usergrid.batch.job" />
<context:annotation-config />
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/99c80faf/stack/pom.xml
----------------------------------------------------------------------
diff --git a/stack/pom.xml b/stack/pom.xml
index adc02fd..600217c 100644
--- a/stack/pom.xml
+++ b/stack/pom.xml
@@ -115,7 +115,7 @@
<antlr.version>3.4</antlr.version>
<tika.version>1.4</tika.version>
<elasticsearch.version>1.2.3</elasticsearch.version>
-
+ <metrics.version>3.0.0</metrics.version>
</properties>
<licenses>
[6/8] adding notificationsservice
Posted by sf...@apache.org.
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/9d7901ae/stack/services/src/main/java/org/apache/usergrid/services/notifications/QueueJob.java
----------------------------------------------------------------------
diff --git a/stack/services/src/main/java/org/apache/usergrid/services/notifications/QueueJob.java b/stack/services/src/main/java/org/apache/usergrid/services/notifications/QueueJob.java
new file mode 100644
index 0000000..554074e
--- /dev/null
+++ b/stack/services/src/main/java/org/apache/usergrid/services/notifications/QueueJob.java
@@ -0,0 +1,127 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.usergrid.services.notifications;
+
+
+import java.util.UUID;
+
+import javax.annotation.PostConstruct;
+
+import org.apache.usergrid.persistence.Notification;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import org.apache.usergrid.batch.JobExecution;
+import org.apache.usergrid.batch.job.OnlyOnceJob;
+import org.apache.usergrid.metrics.MetricsFactory;
+import org.apache.usergrid.persistence.EntityManager;
+import org.apache.usergrid.persistence.EntityManagerFactory;
+import org.apache.usergrid.persistence.entities.JobData;
+import org.apache.usergrid.services.ServiceManager;
+import org.apache.usergrid.services.ServiceManagerFactory;
+
+import com.codahale.metrics.Histogram;
+import com.codahale.metrics.Meter;
+import com.codahale.metrics.Timer;
+
+
+@Component( "queueJob" )
+public class QueueJob extends OnlyOnceJob {
+
+ private static final Logger logger = LoggerFactory.getLogger( QueueJob.class );
+
+ @Autowired
+ private MetricsFactory metricsService;
+
+ @Autowired
+ private ServiceManagerFactory smf;
+
+ @Autowired
+ private EntityManagerFactory emf;
+ private Histogram histogram;
+ private Meter requests;
+ private Timer execution;
+
+
+ public QueueJob() {
+ logger.info( "QueueJob created: " + this );
+ }
+
+
+ @PostConstruct
+ void init() {
+ histogram = metricsService.getHistogram( QueueJob.class, "cycle" );
+ requests = metricsService.getMeter( QueueJob.class, "requests" );
+ execution = metricsService.getTimer( QueueJob.class, "execution" );
+ }
+
+
+ @Override
+ public void doJob( JobExecution jobExecution ) throws Exception {
+ Timer.Context timer = execution.time();
+ requests.mark();
+ logger.info( "execute QueueJob {}", jobExecution );
+
+ JobData jobData = jobExecution.getJobData();
+ UUID applicationId = ( UUID ) jobData.getProperty( "applicationId" );
+ ServiceManager sm = smf.getServiceManager( applicationId );
+ NotificationsService notificationsService = ( NotificationsService ) sm.getService( "notifications" );
+
+ EntityManager em = emf.getEntityManager( applicationId );
+
+ try {
+ if ( em == null ) {
+ logger.info( "no EntityManager for applicationId {}", applicationId );
+ return;
+ }
+ UUID notificationId = ( UUID ) jobData.getProperty( "notificationId" );
+ Notification notification = em.get( notificationId, Notification.class );
+ if ( notification == null ) {
+ logger.info( "notificationId {} no longer exists", notificationId );
+ return;
+ }
+
+ try {
+ notificationsService.getQueueManager().queueNotification( notification, jobExecution );
+ }
+ catch ( Exception e ) {
+ logger.error( "execute QueueJob failed", e );
+ em.setProperty( notification, "errorMessage", e.getMessage() );
+ throw e;
+ }
+ finally {
+ long diff = System.currentTimeMillis() - notification.getCreated();
+ histogram.update( diff );
+ }
+ }
+ finally {
+ timer.stop();
+ }
+
+ logger.info( "execute QueueJob completed normally" );
+ }
+
+
+ @Override
+ protected long getDelay( JobExecution execution ) throws Exception {
+ return TaskManager.BATCH_DEATH_PERIOD;
+ }
+
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/9d7901ae/stack/services/src/main/java/org/apache/usergrid/services/notifications/TaskManager.java
----------------------------------------------------------------------
diff --git a/stack/services/src/main/java/org/apache/usergrid/services/notifications/TaskManager.java b/stack/services/src/main/java/org/apache/usergrid/services/notifications/TaskManager.java
new file mode 100644
index 0000000..e6c876c
--- /dev/null
+++ b/stack/services/src/main/java/org/apache/usergrid/services/notifications/TaskManager.java
@@ -0,0 +1,196 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.usergrid.services.notifications;
+
+import java.util.*;
+
+import org.apache.usergrid.persistence.*;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
+import org.apache.usergrid.mq.Message;
+import org.apache.usergrid.mq.QueueManager;
+import org.apache.usergrid.mq.QueueResults;
+import org.apache.usergrid.persistence.entities.Device;
+
+/**
+ * When all Tasks are "complete", this calls notifyAll(). Note: This may not
+ * mean that all work is done, however, as delivery errors may come in after a
+ * notification is "sent."
+ */
+public class TaskManager {
+
+ // period to poll for batch completion (keep well under Job scheduler
+ // heartbeat value!)
+ private static final long BATCH_POLL_PERIOD = 60 * 1000;
+ // period to tell job scheduler to wait between heartbeats before timing out
+ // this transaction
+ public static final long SCHEDULER_HEARTBEAT_PERIOD = 5 * 60 * 1000;
+ // period at which the batch is considered dead without activity (10
+ // minutes)
+ // setting it high means that a batch that is dead will hang for longer
+ // but setting it too low may cause duplicates to be sent.
+ // also used for Delay before another Job will be attempted - thus total
+ // time
+ // before a job might be restarted could be as long as 2 x
+ // BATCH_DEATH_PERIOD
+ static final long BATCH_DEATH_PERIOD = 10 * 60 * 1000;
+ public static final long MESSAGE_TRANSACTION_TIMEOUT = SCHEDULER_HEARTBEAT_PERIOD;
+
+ private static final Logger LOG = LoggerFactory
+ .getLogger(TaskManager.class);
+ private final String path;
+
+ private Notification notification;
+ private ConcurrentHashMap<UUID,Message> remaining;
+ private AtomicLong successes = new AtomicLong();
+ private AtomicLong failures = new AtomicLong();
+ private AtomicLong skips = new AtomicLong();
+ private QueueManager qm;
+ private EntityManager em;
+ private NotificationServiceProxy ns;
+
+ public TaskManager( EntityManager em, NotificationServiceProxy ns,QueueManager qm, Notification notification, QueueResults queueResults) {
+ this.em = em;
+ this.qm = qm;
+ this.ns = ns;
+ this.path = queueResults.getPath();
+ this.notification = notification;
+ this.remaining = new ConcurrentHashMap<UUID,Message>();
+ for (Message m : queueResults.getMessages()) {
+ remaining
+ .put((UUID) m.getObjectProperty("deviceUUID"), m);
+ }
+ }
+
+ public void skip(UUID deviceUUID) throws Exception {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("notification {} skipped device {}",
+ notification.getUuid(), deviceUUID);
+ }
+ skips.incrementAndGet();
+ completed(null, null, deviceUUID, null);
+ }
+
+ public void completed(Notifier notifier, Receipt receipt, UUID deviceUUID,
+ String newProviderId) throws Exception {
+
+ LOG.debug("REMOVED {}", deviceUUID);
+ try {
+ EntityRef deviceRef = new SimpleEntityRef(Device.ENTITY_TYPE,
+ deviceUUID);
+
+ if (receipt != null) {
+ LOG.debug("notification {} sent to device {}. saving receipt.",
+ notification.getUuid(), deviceUUID);
+ successes.incrementAndGet();
+ receipt.setSent(System.currentTimeMillis());
+ this.saveReceipt(notification, deviceRef, receipt);
+ LOG.debug("notification {} receipt saved for device {}",
+ notification.getUuid(), deviceUUID);
+ }
+
+ if (remaining.containsKey(deviceUUID)) {
+ LOG.debug("notification {} removing device {} from remaining", notification.getUuid(), deviceUUID);
+ qm.commitTransaction(path, remaining.get(deviceUUID).getTransaction(), null);
+ }
+
+ if (newProviderId != null) {
+ LOG.debug("notification {} replacing device {} notifierId", notification.getUuid(), deviceUUID);
+ replaceProviderId(deviceRef, notifier, newProviderId);
+ }
+
+ LOG.debug("notification {} completed device {}", notification.getUuid(), deviceUUID);
+
+ } finally {
+ LOG.debug("COUNT is: {}", successes.get());
+ remaining.remove(deviceUUID);
+ // note: stats are transient for the duration of the batch
+ if (remaining.size() == 0) {
+ long successesCopy = successes.get();
+ long failuresCopy = failures.get();
+ if (successesCopy > 0 || failuresCopy > 0 || skips.get()>0) {
+ ns.finishedBatch(notification, successesCopy, failuresCopy);
+ }
+ }
+
+ }
+ }
+
+ public void failed(Notifier notifier, Receipt receipt, UUID deviceUUID, Object code,
+ String message) throws Exception {
+
+ try {
+
+
+ if (LOG.isDebugEnabled()) {
+ StringBuilder sb = new StringBuilder();
+ sb.append("notification ").append(notification.getUuid());
+ sb.append(" for device ").append(deviceUUID);
+ sb.append(" got error ").append(code);
+ LOG.debug(sb.toString());
+ }
+
+ failures.incrementAndGet();
+ if (receipt.getUuid() != null) {
+ successes.decrementAndGet();
+ }
+ receipt.setErrorCode(code);
+ receipt.setErrorMessage(message);
+ this.saveReceipt(notification, new SimpleEntityRef( Device.ENTITY_TYPE, deviceUUID), receipt);
+ LOG.debug("notification {} receipt saved for device {}", notification.getUuid(), deviceUUID);
+ } finally {
+ completed(notifier, null, deviceUUID, null);
+ }
+ }
+
+ /*
+ * called from TaskManager - creates a persistent receipt and updates the
+ * passed one w/ the UUID
+ */
+ public void saveReceipt(EntityRef notification, EntityRef device,
+ Receipt receipt) throws Exception {
+ if (receipt.getUuid() == null) {
+ Receipt savedReceipt = em.create(receipt);
+ receipt.setUuid(savedReceipt.getUuid());
+
+ List<EntityRef> entities = Arrays.asList(notification, device);
+ em.addToCollections(entities, Notification.RECEIPTS_COLLECTION, savedReceipt);
+ } else {
+ em.update(receipt);
+ }
+ }
+
+ protected void replaceProviderId(EntityRef device, Notifier notifier,
+ String newProviderId) throws Exception {
+ Object value = em.getProperty(device, notifier.getName()
+ + NotificationsService.NOTIFIER_ID_POSTFIX);
+ if (value != null) {
+ em.setProperty(device, notifier.getName() + NotificationsService.NOTIFIER_ID_POSTFIX, newProviderId);
+ } else {
+ value = em.getProperty(device, notifier.getUuid()
+ + NotificationsService.NOTIFIER_ID_POSTFIX);
+ if (value != null) {
+ em.setProperty(device, notifier.getUuid() + NotificationsService.NOTIFIER_ID_POSTFIX, newProviderId);
+ }
+ }
+ }
+
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/9d7901ae/stack/services/src/main/java/org/apache/usergrid/services/notifications/TaskTracker.java
----------------------------------------------------------------------
diff --git a/stack/services/src/main/java/org/apache/usergrid/services/notifications/TaskTracker.java b/stack/services/src/main/java/org/apache/usergrid/services/notifications/TaskTracker.java
new file mode 100644
index 0000000..3b4d3d7
--- /dev/null
+++ b/stack/services/src/main/java/org/apache/usergrid/services/notifications/TaskTracker.java
@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.usergrid.services.notifications;
+
+
+import org.apache.usergrid.persistence.Notifier;
+import org.apache.usergrid.persistence.Receipt;
+
+import java.util.UUID;
+
+public class TaskTracker {
+
+ private Notifier notifier;
+ private TaskManager taskManager;
+ private Receipt receipt;
+ private UUID id;
+
+ public TaskTracker(Notifier notifier, TaskManager taskManager, Receipt receipt, UUID id) {
+ this.notifier = notifier;
+ this.taskManager = taskManager;
+ this.receipt = receipt;
+ this.id = id;
+ }
+
+ public void completed() throws Exception {
+ taskManager.completed(notifier, receipt, id, null);
+ }
+
+ public void failed(Object code, String message) throws Exception {
+ taskManager.failed(notifier, receipt, id, code, message);
+ }
+
+ public void completed(String newToken) throws Exception {
+ taskManager.completed(notifier, receipt, id, newToken);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/9d7901ae/stack/services/src/main/java/org/apache/usergrid/services/notifications/TestAdapter.java
----------------------------------------------------------------------
diff --git a/stack/services/src/main/java/org/apache/usergrid/services/notifications/TestAdapter.java b/stack/services/src/main/java/org/apache/usergrid/services/notifications/TestAdapter.java
new file mode 100644
index 0000000..b3f6243
--- /dev/null
+++ b/stack/services/src/main/java/org/apache/usergrid/services/notifications/TestAdapter.java
@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.usergrid.services.notifications;
+
+import org.apache.usergrid.persistence.Notification;
+import org.apache.usergrid.persistence.Notifier;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+import java.util.Date;
+import java.util.Map;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import org.apache.usergrid.persistence.EntityManager;
+import org.apache.usergrid.services.ServicePayload;
+import org.apache.usergrid.services.notifications.apns.APNsAdapter;
+import org.apache.usergrid.services.notifications.apns.APNsNotification;
+
+/**
+ * Just used for testing. Performance and such.
+ */
+public class TestAdapter implements ProviderAdapter {
+
+ private static final Logger log = LoggerFactory.getLogger(TestAdapter.class);
+ private static final int DELAY = 1; // if delay > 0, uses threadpool
+
+ private ExecutorService pool = null;
+
+ public TestAdapter() {
+ if (DELAY > 0) {
+ pool = Executors
+ .newFixedThreadPool(APNsAdapter.MAX_CONNECTION_POOL_SIZE);
+ }
+ }
+
+ @Override
+ public void testConnection(Notifier notifier) throws ConnectionException {
+ }
+
+ @Override
+ public void sendNotification(
+ String providerId,
+ Notifier notifier,
+ final Object payload,
+ Notification notification,
+ TaskTracker tracker)
+ throws Exception {
+
+ final APNsNotification apnsNotification = APNsNotification.create(
+ "", payload.toString(), notification, tracker);
+
+ if (pool == null) {
+ apnsNotification.messageSent();
+
+ } else {
+ pool.submit(new Runnable() {
+ @Override
+ public void run() {
+ try {
+ Thread.sleep(DELAY);
+ apnsNotification.messageSent();
+ log.debug("messageSent() - " + payload.toString());
+ } catch (Exception e) {
+ log.error("messageSent() returned error", e);
+ }
+ }
+ });
+ }
+ }
+
+ @Override
+ public void doneSendingNotifications() throws Exception {
+ log.debug("doneSendingNotifications()");
+ }
+
+ @Override
+ public Map<String, Date> getInactiveDevices(Notifier notifier,
+ EntityManager em) throws Exception {
+ log.debug("getInactiveDevices()");
+ return null;
+ }
+
+ @Override
+ public Object translatePayload(Object payload) throws Exception {
+ return payload;
+ }
+
+ @Override
+ public void validateCreateNotifier(ServicePayload payload) throws Exception {
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/9d7901ae/stack/services/src/main/java/org/apache/usergrid/services/notifications/apns/APNsAdapter.java
----------------------------------------------------------------------
diff --git a/stack/services/src/main/java/org/apache/usergrid/services/notifications/apns/APNsAdapter.java b/stack/services/src/main/java/org/apache/usergrid/services/notifications/apns/APNsAdapter.java
new file mode 100644
index 0000000..73399c9
--- /dev/null
+++ b/stack/services/src/main/java/org/apache/usergrid/services/notifications/apns/APNsAdapter.java
@@ -0,0 +1,280 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.usergrid.services.notifications.apns;
+
+import com.google.common.cache.*;
+
+import com.relayrides.pushy.apns.*;
+import com.relayrides.pushy.apns.util.*;
+
+import org.apache.usergrid.persistence.Notification;
+import org.apache.usergrid.persistence.Notifier;
+import org.mortbay.util.ajax.JSON;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.InputStream;
+import java.security.*;
+import java.util.*;
+import java.util.concurrent.*;
+
+import org.apache.usergrid.persistence.EntityManager;
+import org.apache.usergrid.persistence.exceptions.RequiredPropertyNotFoundException;
+import org.apache.usergrid.services.ServicePayload;
+import org.apache.usergrid.services.notifications.ConnectionException;
+import org.apache.usergrid.services.notifications.ProviderAdapter;
+import org.apache.usergrid.services.notifications.TaskTracker;
+
+import javax.net.ssl.SSLContext;
+
+/**
+ * Adapter for Apple push notifications
+ */
+public class APNsAdapter implements ProviderAdapter {
+
+ private static final Logger logger = LoggerFactory
+ .getLogger(APNsAdapter.class);
+
+ public static int MAX_CONNECTION_POOL_SIZE = 15;
+ private static final Set<String> validEnvironments = new HashSet<String>();
+ private static final String TEST_TOKEN = "ff026b5a4d2761ef13843e8bcab9fc83b47f1dfbd1d977d225ab296153ce06d6";
+ private static final String TEST_PAYLOAD = "{}";
+
+ static {
+ validEnvironments.add("development");
+ validEnvironments.add("production");
+ validEnvironments.add("mock");
+ }
+
+ public APNsAdapter(){}
+
+ @Override
+ public void testConnection(Notifier notifier) throws ConnectionException {
+ if(isMock(notifier)){
+ delayRandom(notifier); return;
+ }
+ TestAPNsNotification notification = TestAPNsNotification.create(TEST_TOKEN, TEST_PAYLOAD);
+ try {
+ CountDownLatch latch = new CountDownLatch(1);
+ notification.setLatch(latch);
+ PushManager<SimpleApnsPushNotification> pushManager = getPushManager(notifier);
+ addToQueue(pushManager, notification);
+ latch.await(10000,TimeUnit.MILLISECONDS);
+ if(notification.hasFailed()){
+ // this is expected with a bad certificate (w/message: comes from failedconnectionlistener
+ throw new ConnectionException("Bad certificate. Double-check your environment.",notification.getCause() != null ? notification.getCause() : new Exception("Bad certificate."));
+ }
+ notification.finished();
+ } catch (Exception e) {
+ notification.finished();
+
+ if (e instanceof ConnectionException) {
+ throw (ConnectionException) e;
+ }
+ if (e instanceof InterruptedException) {
+ throw new ConnectionException("Test notification timed out", e);
+ }
+ logger.warn("testConnection got non-fatal error", e.getCause());
+ }
+ }
+
+ private BlockingQueue<SimpleApnsPushNotification> addToQueue(PushManager<SimpleApnsPushNotification> pushManager, SimpleApnsPushNotification notification) throws InterruptedException {
+ BlockingQueue<SimpleApnsPushNotification> queue = pushManager.getQueue();
+ queue.offer(notification,2500,TimeUnit.MILLISECONDS);
+ return queue;
+ }
+
+ @Override
+ public void sendNotification(String providerId, Notifier notifier,
+ Object payload, Notification notification, TaskTracker tracker)
+ throws Exception {
+ if(isMock(notifier)){
+ tracker.completed("Mocked!");
+ return;
+ }
+ APNsNotification apnsNotification = APNsNotification.create(providerId, payload.toString(), notification, tracker);
+ PushManager<SimpleApnsPushNotification> pushManager = getPushManager(notifier);
+ try {
+ addToQueue(pushManager, apnsNotification);
+ apnsNotification.messageSent();
+ }catch (InterruptedException ie){
+ apnsNotification.messageSendFailed(ie);
+ throw ie;
+ }
+ }
+
+ @Override
+ public void doneSendingNotifications() throws Exception {
+ // do nothing - no batching
+ }
+
+ @Override
+ public Map<String, Date> getInactiveDevices(Notifier notifier,
+ EntityManager em) throws Exception {
+ Map<String,Date> map = new HashMap<String,Date>();
+ if(isMock(notifier)){
+ return map;
+ }
+ PushManager<SimpleApnsPushNotification> pushManager = getPushManager(notifier);
+
+ List<ExpiredToken> tokens = null;
+ try {
+ tokens = pushManager.getExpiredTokens();
+ }catch (FeedbackConnectionException fce){
+ logger.debug("Failed to get tokens",fce);
+ return map;
+ }
+ for(ExpiredToken token : tokens){
+ String expiredToken = new String(token.getToken());
+ map.put(expiredToken, token.getExpiration());
+ }
+ return map;
+ }
+
+ private PushManager<SimpleApnsPushNotification> getPushManager(Notifier notifier) throws ExecutionException {
+ PushManager<SimpleApnsPushNotification> pushManager = apnsServiceMap.get(notifier);
+ if(pushManager != null && !pushManager.isStarted() && pushManager.isShutDown()){
+ try{
+ pushManager = createApnsService(notifier);
+ }catch(Exception e){
+ logger.error("could not instantiate push manager.");
+ throw new ExecutionException(e);
+ }
+ apnsServiceMap.put(notifier,pushManager);
+ }
+ try {
+ if (!pushManager.isStarted()) { //ensure manager is started
+ pushManager.start();
+ }
+ }catch(IllegalStateException ise){
+ logger.debug("failed to start",ise);//could have failed because its starteded
+ }
+ return pushManager;
+ }
+
+ //cache to retrieve push manager, cached per notifier, so many notifications will get same push manager
+ private static LoadingCache<Notifier, PushManager<SimpleApnsPushNotification>> apnsServiceMap = CacheBuilder
+ .newBuilder().expireAfterAccess(10, TimeUnit.MINUTES)
+ .removalListener(new RemovalListener<Notifier, PushManager<SimpleApnsPushNotification>>() {
+ @Override
+ public void onRemoval(
+ RemovalNotification<Notifier, PushManager<SimpleApnsPushNotification>> notification) {
+ try {
+ PushManager<SimpleApnsPushNotification> manager = notification.getValue();
+ if(!manager.isShutDown()){
+ notification.getValue().shutdown();
+ }
+ } catch (Exception ie) {
+ logger.error("Failed to shutdown from cache",ie);
+ }
+ }
+ }).build(new CacheLoader<Notifier, PushManager<SimpleApnsPushNotification>>() {
+ @Override
+ public PushManager<SimpleApnsPushNotification> load(Notifier notifier) {
+ try{
+ return createApnsService(notifier);
+ }catch (KeyStoreException ke){
+ logger.error("Could not instantiate pushmanager",ke);
+ return null;
+ }
+ }
+ });
+
+
+ protected static PushManager<SimpleApnsPushNotification> createApnsService(Notifier notifier) throws KeyStoreException{
+ LinkedBlockingQueue<SimpleApnsPushNotification> queue = new LinkedBlockingQueue<SimpleApnsPushNotification>();
+ PushManagerConfiguration config = new PushManagerConfiguration();
+ config.setConcurrentConnectionCount(Runtime.getRuntime().availableProcessors() * 2);
+ PushManager<SimpleApnsPushNotification> pushManager = new PushManager<SimpleApnsPushNotification>(getApnsEnvironment(notifier), getSSLContext(notifier), null, null, queue, config);
+ //only tested when a message is sent
+ pushManager.registerRejectedNotificationListener(new RejectedAPNsListener());
+ //this will get tested when start is called
+ pushManager.registerFailedConnectionListener(new FailedConnectionListener());
+ return pushManager;
+ }
+
+ @Override
+ public Object translatePayload(Object objPayload) throws Exception {
+ String payload;
+ if (objPayload instanceof String) {
+ payload = (String) objPayload;
+ if (!payload.startsWith("{")) {
+ payload = "{\"aps\":{\"alert\":\"" + payload + "\"}}";
+ }
+ } else {
+ payload = JSON.toString(objPayload);
+ }
+ if (payload.length() > 256) {
+ throw new IllegalArgumentException(
+ "Apple APNs payloads must be 256 characters or less");
+ }
+ return payload;
+ }
+
+
+ @Override
+ public void validateCreateNotifier(ServicePayload payload) throws Exception {
+ String environment = payload.getStringProperty("environment");
+ if (!validEnvironments.contains(environment)) {
+ throw new IllegalArgumentException("environment must be one of: "
+ + Arrays.toString(validEnvironments.toArray()));
+ }
+
+ if (payload.getProperty("p12Certificate") == null) {
+ throw new RequiredPropertyNotFoundException("notifier",
+ "p12Certificate");
+ }
+ }
+ public boolean isMock(Notifier notifier){
+ return notifier.getEnvironment() !=null ? notifier.getEnvironment().equals("mock") : false ;
+ }
+ public boolean delayRandom(Notifier notifier) {
+ boolean wasDelayed = false;
+ if (isMock(notifier)) {
+ try {
+ Thread.sleep(
+ new Random().nextInt(300)
+ );
+ wasDelayed = true;
+ } catch (InterruptedException ie) {
+ //delay was stopped
+ }
+ }
+ return wasDelayed;
+ }
+
+ private static ApnsEnvironment getApnsEnvironment(Notifier notifier){
+ return notifier.isProduction()
+ ? ApnsEnvironment.getProductionEnvironment()
+ : ApnsEnvironment.getSandboxEnvironment();
+ }
+
+
+ private static SSLContext getSSLContext(Notifier notifier) {
+ try {
+ KeyStore keyStore = KeyStore.getInstance("PKCS12");
+ String password = notifier.getCertificatePassword();
+ char[] passChars =(password != null ? password : "").toCharArray();
+ InputStream stream = notifier.getP12CertificateStream();
+ keyStore.load(stream,passChars);
+ SSLContext context = SSLContextUtil.createDefaultSSLContext(keyStore, passChars);
+ return context;
+ }catch (Exception e){
+ throw new RuntimeException("Error getting certificate",e);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/9d7901ae/stack/services/src/main/java/org/apache/usergrid/services/notifications/apns/APNsNotification.java
----------------------------------------------------------------------
diff --git a/stack/services/src/main/java/org/apache/usergrid/services/notifications/apns/APNsNotification.java b/stack/services/src/main/java/org/apache/usergrid/services/notifications/apns/APNsNotification.java
new file mode 100644
index 0000000..11a9856
--- /dev/null
+++ b/stack/services/src/main/java/org/apache/usergrid/services/notifications/apns/APNsNotification.java
@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.usergrid.services.notifications.apns;
+
+import com.relayrides.pushy.apns.RejectedNotificationReason;
+import com.relayrides.pushy.apns.util.MalformedTokenStringException;
+import com.relayrides.pushy.apns.util.SimpleApnsPushNotification;
+import com.relayrides.pushy.apns.util.TokenUtil;
+
+import org.apache.usergrid.persistence.Notification;
+import org.apache.usergrid.services.notifications.TaskTracker;
+
+import java.util.Calendar;
+import java.util.Date;
+/**
+ * Standard apigee notificatton
+ */
+public class APNsNotification extends SimpleApnsPushNotification {
+
+
+ private TaskTracker tracker;
+
+ /**
+ * Factory method
+ * @param providerId token for device
+ * @param payload body
+ * @param notification notification entity
+ * @param tracker tracks completion
+ * @return
+ */
+ public static APNsNotification create(String providerId, String payload, Notification notification, TaskTracker tracker) throws RuntimeException {
+
+ Calendar date = Calendar.getInstance();
+ date.add(Calendar.SECOND, notification.getExpireTimeInSeconds());
+ try {
+ final byte[] token = TokenUtil.tokenStringToByteArray(providerId);
+
+ return new APNsNotification(tracker, date.getTime(), token, payload, notification);
+ }catch(MalformedTokenStringException mtse){
+ throw new RuntimeException("Exception converting token",mtse);
+ }
+ }
+
+ /**
+ * Default constructor
+ * @param tracker
+ * @param expiryTime
+ * @param token
+ * @param payload
+ */
+ public APNsNotification(TaskTracker tracker, Date expiryTime, byte[] token, String payload,Notification notification) {
+ super(token, payload, expiryTime);
+ this.tracker = tracker;
+ }
+
+ /**
+ * mark message sent
+ * @throws Exception
+ */
+ public void messageSent() throws Exception {
+ if (tracker != null) {
+ tracker.completed();
+ }
+ }
+
+ /**
+ * mark message failed
+ *
+ * @throws Exception
+ */
+ public void messageSendFailed(RejectedNotificationReason reason) throws Exception {
+ if (tracker != null) {
+ tracker.failed(reason.name(), "Failed sending notification.");
+ }
+ }
+
+ /**
+ * mark message failed, from exception
+ * @param cause
+ * @throws Exception
+ */
+ public void messageSendFailed(Throwable cause) throws Exception {
+ if (tracker != null) {
+ tracker.failed(cause.getClass().getSimpleName(), cause.getMessage());
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/9d7901ae/stack/services/src/main/java/org/apache/usergrid/services/notifications/apns/FailedConnectionListener.java
----------------------------------------------------------------------
diff --git a/stack/services/src/main/java/org/apache/usergrid/services/notifications/apns/FailedConnectionListener.java b/stack/services/src/main/java/org/apache/usergrid/services/notifications/apns/FailedConnectionListener.java
new file mode 100644
index 0000000..552701f
--- /dev/null
+++ b/stack/services/src/main/java/org/apache/usergrid/services/notifications/apns/FailedConnectionListener.java
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.usergrid.services.notifications.apns;
+
+import com.relayrides.pushy.apns.*;
+import com.relayrides.pushy.apns.util.SimpleApnsPushNotification;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.net.ssl.SSLException;
+import javax.net.ssl.SSLHandshakeException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+/**
+ * Provides a single listener that basically just delegates back to the
+ * APNsNotification for handling.
+ */
+public class FailedConnectionListener implements com.relayrides.pushy.apns.FailedConnectionListener<SimpleApnsPushNotification> {
+
+ private static final Logger logger = LoggerFactory
+ .getLogger(RejectedAPNsListener.class);
+
+ @Override
+ public void handleFailedConnection(PushManager<? extends SimpleApnsPushNotification> pushManager, Throwable cause) {
+ List<SimpleApnsPushNotification> notifications = new ArrayList<SimpleApnsPushNotification>();
+ if (cause instanceof SSLException || cause instanceof SSLHandshakeException) { //cert is probably bad so shut it down.
+ if (!pushManager.isShutDown()) {
+ pushManager.unregisterFailedConnectionListener(this);
+
+ try {
+ BlockingQueue notificationQueue = pushManager.getQueue();
+ if(notificationQueue !=null){
+ LinkedBlockingQueue<SimpleApnsPushNotification> queue = ( LinkedBlockingQueue<SimpleApnsPushNotification> )notificationQueue;
+ Object[] objectMess = queue.toArray(); //get messages still in queue
+ for(Object o : objectMess){
+ if(o instanceof SimpleApnsPushNotification) {
+ notifications.add((SimpleApnsPushNotification) o);
+ }
+ }
+ }
+ pushManager.shutdown();
+ } catch (InterruptedException ie) {
+ logger.error("Failed to stop push services", ie);
+ }
+ } else {
+ return;
+ }
+ }
+ //mark all unsent notifications failed
+ if (notifications != null) {
+ for (SimpleApnsPushNotification notification : notifications) {
+ if (notification instanceof APNsNotification) {
+ try {
+ ((APNsNotification) notification).messageSendFailed(cause);//mark failed with bad token
+ } catch (Exception e) {
+ logger.error("failed to track notification in failed connection listener", e);
+ }
+ }
+ //if test this is a problem because you can't connect
+ if (notification instanceof TestAPNsNotification) {
+ TestAPNsNotification testAPNsNotification = ((TestAPNsNotification) notification);
+ testAPNsNotification.setReason(cause);
+ testAPNsNotification.countdown();
+ }
+
+ }
+ pushManager.getQueue().clear();
+ }
+ logger.error("Failed to register push connection", cause);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/9d7901ae/stack/services/src/main/java/org/apache/usergrid/services/notifications/apns/RejectedAPNsListener.java
----------------------------------------------------------------------
diff --git a/stack/services/src/main/java/org/apache/usergrid/services/notifications/apns/RejectedAPNsListener.java b/stack/services/src/main/java/org/apache/usergrid/services/notifications/apns/RejectedAPNsListener.java
new file mode 100644
index 0000000..545dc0c
--- /dev/null
+++ b/stack/services/src/main/java/org/apache/usergrid/services/notifications/apns/RejectedAPNsListener.java
@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.usergrid.services.notifications.apns;
+
+import com.relayrides.pushy.apns.*;
+import com.relayrides.pushy.apns.util.*;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Provides a single listener that basically just delegates back to the
+ * APNsNotification for handling.
+ */
+public class RejectedAPNsListener implements RejectedNotificationListener<SimpleApnsPushNotification>{
+
+ @Override
+ public void handleRejectedNotification(PushManager<? extends SimpleApnsPushNotification> pushManager, SimpleApnsPushNotification notification, RejectedNotificationReason rejectionReason) {
+ try {
+ //mark failed for standard notification
+ if (notification instanceof APNsNotification) {
+ ((APNsNotification) notification).messageSendFailed(rejectionReason);
+ }
+ //if test getting here means it worked
+ if(notification instanceof TestAPNsNotification){
+ TestAPNsNotification testAPNsNotification = (TestAPNsNotification) notification;
+ testAPNsNotification.setReason(rejectionReason);
+ testAPNsNotification.countdown();
+ logger.error("Failed to connect to APN's service",testAPNsNotification);
+ }
+
+ } catch (Exception e) {
+ logger.error("Failed to track rejected listener", e);
+ }
+ System.out.format("%s was rejected with rejection reason %s\n", notification, rejectionReason);
+ }
+
+ private static final Logger logger = LoggerFactory.getLogger(RejectedAPNsListener.class);
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/9d7901ae/stack/services/src/main/java/org/apache/usergrid/services/notifications/apns/TestAPNsListener.java
----------------------------------------------------------------------
diff --git a/stack/services/src/main/java/org/apache/usergrid/services/notifications/apns/TestAPNsListener.java b/stack/services/src/main/java/org/apache/usergrid/services/notifications/apns/TestAPNsListener.java
new file mode 100644
index 0000000..9d112d4
--- /dev/null
+++ b/stack/services/src/main/java/org/apache/usergrid/services/notifications/apns/TestAPNsListener.java
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.usergrid.services.notifications.apns;
+
+import com.relayrides.pushy.apns.ApnsConnection;
+import com.relayrides.pushy.apns.ApnsConnectionListener;
+import com.relayrides.pushy.apns.RejectedNotificationReason;
+import com.relayrides.pushy.apns.util.SimpleApnsPushNotification;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.concurrent.CountDownLatch;
+
+public class TestAPNsListener implements ApnsConnectionListener<SimpleApnsPushNotification> {
+
+ private final CountDownLatch latch;
+
+ private boolean connectionFailed = false;
+ private boolean connectionClosed = false;
+
+ private Throwable connectionFailureCause;
+
+ private final ArrayList<SimpleApnsPushNotification> writeFailures = new ArrayList<SimpleApnsPushNotification>();
+
+ private SimpleApnsPushNotification rejectedNotification;
+ private RejectedNotificationReason rejectionReason;
+
+ private final ArrayList<SimpleApnsPushNotification> unprocessedNotifications = new ArrayList<SimpleApnsPushNotification>();
+
+ public TestAPNsListener() {
+ this.latch = new CountDownLatch(1);
+ }
+
+ public void handleConnectionSuccess(final ApnsConnection<SimpleApnsPushNotification> connection) {
+ latch.countDown();
+ }
+
+ public void handleConnectionFailure(final ApnsConnection<SimpleApnsPushNotification> connection, final Throwable cause) {
+ this.connectionFailed = true;
+ this.connectionFailureCause = cause;
+ latch.countDown();
+ }
+
+ public void handleConnectionClosure(ApnsConnection<SimpleApnsPushNotification> connection) {
+ try {
+ connection.waitForPendingWritesToFinish();
+ } catch (InterruptedException ignored) {
+ }
+ this.connectionClosed = true;
+ latch.countDown();
+ }
+
+ public void handleWriteFailure(ApnsConnection<SimpleApnsPushNotification> connection,
+ SimpleApnsPushNotification notification, Throwable cause) {
+
+ this.writeFailures.add(notification);
+ }
+
+ public void handleRejectedNotification(ApnsConnection<SimpleApnsPushNotification> connection,
+ SimpleApnsPushNotification rejectedNotification, RejectedNotificationReason reason) {
+
+ this.rejectedNotification = rejectedNotification;
+ this.rejectionReason = reason;
+ }
+
+ public void handleUnprocessedNotifications(ApnsConnection<SimpleApnsPushNotification> connection,
+ Collection<SimpleApnsPushNotification> unprocessedNotifications) {
+
+ this.unprocessedNotifications.addAll(unprocessedNotifications);
+ }
+
+ public void handleConnectionWritabilityChange(ApnsConnection<SimpleApnsPushNotification> connection, boolean writable) {
+ }
+
+ public boolean hasConnectionFailed() {
+ return connectionFailed;
+ }
+
+ public Throwable getConnectionFailureCause(){
+ return connectionFailureCause;
+ }
+
+ public CountDownLatch getLatch() {
+ return latch;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/9d7901ae/stack/services/src/main/java/org/apache/usergrid/services/notifications/apns/TestAPNsNotification.java
----------------------------------------------------------------------
diff --git a/stack/services/src/main/java/org/apache/usergrid/services/notifications/apns/TestAPNsNotification.java b/stack/services/src/main/java/org/apache/usergrid/services/notifications/apns/TestAPNsNotification.java
new file mode 100644
index 0000000..014d275
--- /dev/null
+++ b/stack/services/src/main/java/org/apache/usergrid/services/notifications/apns/TestAPNsNotification.java
@@ -0,0 +1,124 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.usergrid.services.notifications.apns;
+
+import com.relayrides.pushy.apns.RejectedNotificationReason;
+import com.relayrides.pushy.apns.util.MalformedTokenStringException;
+import com.relayrides.pushy.apns.util.SimpleApnsPushNotification;
+import com.relayrides.pushy.apns.util.TokenUtil;
+import com.yammer.metrics.Metrics;
+import com.yammer.metrics.core.Timer;
+import com.yammer.metrics.core.TimerContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Calendar;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * notification type for testing connections
+ */
+public class TestAPNsNotification extends SimpleApnsPushNotification {
+
+ private static final Logger logger = LoggerFactory.getLogger(TestAPNsNotification.class);
+
+ boolean hasFailed = false;
+
+ CountDownLatch latch;
+
+ private final Timer processTimer =
+ Metrics.newTimer(TestAPNsNotification.class, "apns_test_notification", TimeUnit.MICROSECONDS, TimeUnit.SECONDS);
+ private TimerContext timer;
+ private Throwable cause;
+
+ public static TestAPNsNotification create(String tokenString, String payload) throws RuntimeException{
+ try {
+ final byte[] token = TokenUtil.tokenStringToByteArray(tokenString);
+ return new TestAPNsNotification( token, payload);
+ }catch (MalformedTokenStringException mtse) {
+ throw new RuntimeException("exception foreign byte array",mtse);
+ }
+ }
+
+ /**
+ * setup concurrency
+ * @param latch
+ */
+ public void setLatch(CountDownLatch latch){
+ this.latch = latch;
+ }
+
+ /**
+ * get concurrency
+ * @return
+ */
+ public CountDownLatch getLatch(){
+ return latch;
+ }
+
+ /**
+ * decrement countdown for concurrency
+ */
+ public void countdown(){
+ if(latch != null){
+ latch.countDown();
+ }
+ }
+
+ public TestAPNsNotification( byte[] token, String payload) {
+ super(token, payload, Calendar.getInstance().getTime());
+ this.timer = processTimer.time();
+ }
+
+ /**
+ * has this failed
+ * @return
+ */
+ public boolean hasFailed(){
+ return hasFailed;
+ }
+
+ /**
+ * stop timer
+ */
+ public void finished(){
+ this.timer.stop();
+ }
+
+ /**
+ * get failure reason
+ * @return cause
+ */
+ public Throwable getCause(){return cause;}
+ /**
+ * mark failure state
+ * @param cause
+ */
+ public void setReason(Throwable cause){
+ hasFailed = true; //token is definitely invalid, so don't fail
+ this.cause = cause;
+ }
+ /**
+ * mark failure state
+ * @param reason
+ */
+ public void setReason(RejectedNotificationReason reason){
+ hasFailed = reason != RejectedNotificationReason.INVALID_TOKEN; //token is definitely invalid, so don't fail
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/9d7901ae/stack/services/src/main/java/org/apache/usergrid/services/notifications/gcm/GCMAdapter.java
----------------------------------------------------------------------
diff --git a/stack/services/src/main/java/org/apache/usergrid/services/notifications/gcm/GCMAdapter.java b/stack/services/src/main/java/org/apache/usergrid/services/notifications/gcm/GCMAdapter.java
new file mode 100644
index 0000000..04a07d8
--- /dev/null
+++ b/stack/services/src/main/java/org/apache/usergrid/services/notifications/gcm/GCMAdapter.java
@@ -0,0 +1,221 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.usergrid.services.notifications.gcm;
+
+import com.google.android.gcm.server.*;
+import org.apache.usergrid.persistence.Notification;
+import org.apache.usergrid.persistence.Notifier;
+import org.mortbay.util.ajax.JSON;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.usergrid.services.ServicePayload;
+import org.apache.usergrid.persistence.EntityManager;
+import org.apache.usergrid.persistence.exceptions.RequiredPropertyNotFoundException;
+
+import org.apache.usergrid.services.notifications.ConnectionException;
+import org.apache.usergrid.services.notifications.ProviderAdapter;
+import org.apache.usergrid.services.notifications.TaskTracker;
+
+import java.io.IOException;
+import java.util.*;
+
+public class GCMAdapter implements ProviderAdapter {
+
+ private static final Logger LOG = LoggerFactory.getLogger(GCMAdapter.class);
+ private static final int SEND_RETRIES = 3;
+ private static int BATCH_SIZE = 1000;
+
+ private Map<Notifier, Batch> notifierBatches = new HashMap<Notifier, Batch>();
+
+ @Override
+ public void testConnection(Notifier notifier) throws ConnectionException {
+ if(isMock(notifier)){
+ try{Thread.sleep(200);}catch (Exception ie){}
+ return;
+ }
+ Sender sender = new Sender(notifier.getApiKey());
+ Message message = new Message.Builder().build();
+ try {
+ Result result = sender.send(message, "device_token", 1);
+ LOG.debug("testConnection result: {}", result);
+ } catch (IOException e) {
+ throw new ConnectionException(e.getMessage(), e);
+ }
+ }
+
+ @Override
+ public void sendNotification(String providerId, Notifier notifier,
+ Object payload, Notification notification, TaskTracker tracker)
+ throws Exception {
+ Map<String,Object> map = (Map<String, Object>) payload;
+ final String expiresKey = "time_to_live";
+ if(!map.containsKey(expiresKey) && notification.getExpire() != null){
+ int expireSeconds = notification.getExpireTimeInSeconds();
+ expireSeconds = expireSeconds <= 2419200 ? expireSeconds : 2419200; //send the max gcm value documented here http://developer.android.com/google/gcm/adv.html#ttl
+ map.put(expiresKey, expireSeconds);
+ }
+ Batch batch = getBatch(notifier, map);
+ batch.add(providerId, tracker);
+ }
+
+ synchronized private Batch getBatch(Notifier notifier,
+ Map<String, Object> payload) {
+ Batch batch = notifierBatches.get(notifier);
+ if (batch == null && payload != null) {
+ batch = new Batch(notifier, payload);
+ notifierBatches.put(notifier, batch);
+ }
+ return batch;
+ }
+
+ @Override
+ synchronized public void doneSendingNotifications() throws Exception {
+ for (Batch batch : notifierBatches.values()) {
+ batch.send();
+ }
+ }
+
+ @Override
+ public Map<String, Date> getInactiveDevices(Notifier notifier,
+ EntityManager em) throws Exception {
+ Batch batch = getBatch(notifier, null);
+ Map<String,Date> map = null;
+ if(batch != null) {
+ map = batch.getAndClearInactiveDevices();
+ }
+ return map;
+ }
+
+ @Override
+ public Map<String, Object> translatePayload(Object payload)
+ throws Exception {
+ Map<String, Object> mapPayload = new HashMap<String, Object>();
+ if (payload instanceof Map) {
+ mapPayload = (Map<String, Object>) payload;
+ } else if (payload instanceof String) {
+ mapPayload.put("data", payload);
+ } else {
+ throw new IllegalArgumentException(
+ "GCM Payload must be either a Map or a String");
+ }
+ if (JSON.toString(mapPayload).length() > 4096) {
+ throw new IllegalArgumentException(
+ "GCM payloads must be 4096 characters or less");
+ }
+ return mapPayload;
+ }
+
+ @Override
+ public void validateCreateNotifier(ServicePayload payload) throws Exception {
+ if (payload.getProperty("apiKey") == null) {
+ throw new RequiredPropertyNotFoundException("notifier", "apiKey");
+ }
+ }
+
+ private class Batch {
+ private Notifier notifier;
+ private Map payload;
+ private List<String> ids;
+ private List<TaskTracker> trackers;
+ private Map<String, Date> inactiveDevices = new HashMap<String, Date>();
+
+ Batch(Notifier notifier, Map<String, Object> payload) {
+ this.notifier = notifier;
+ this.payload = payload;
+ this.ids = new ArrayList<String>();
+ this.trackers = new ArrayList<TaskTracker>();
+ }
+
+ synchronized Map<String, Date> getAndClearInactiveDevices() {
+ Map<String, Date> map = inactiveDevices;
+ inactiveDevices = new HashMap<String, Date>();
+ return map;
+ }
+
+ synchronized void add(String id, TaskTracker tracker) throws Exception {
+ ids.add(id);
+ trackers.add(tracker);
+
+ if (ids.size() == BATCH_SIZE) {
+ send();
+ }
+ }
+
+ // Message.Builder requires the payload to be Map<String,String> for no
+ // good reason, so I just blind cast it.
+ // What actually happens is: "JSONValue.toJSONString(payload);" so
+ // anything that JSONValue can handle is fine.
+ // (What is necessary here is that the Map needs to have a nested
+ // structure.)
+ synchronized void send() throws Exception {
+ if (ids.size() == 0)
+ return;
+ Sender sender = new Sender(notifier.getApiKey());
+ Message.Builder builder = new Message.Builder();
+ builder.setData(payload);
+ Message message = builder.build();
+ if(isMock(notifier)){
+ delayRandom(notifier);
+ for(TaskTracker tracker : trackers){
+ tracker.completed("Mocked!");
+ }
+ return;
+ }else {
+ MulticastResult multicastResult = sender.send(message, ids,
+ SEND_RETRIES);
+ LOG.debug("sendNotification result: {}", multicastResult);
+
+ for (int i = 0; i < multicastResult.getTotal(); i++) {
+ Result result = multicastResult.getResults().get(i);
+
+ if (result.getMessageId() != null) {
+ String canonicalRegId = result.getCanonicalRegistrationId();
+ trackers.get(i).completed(canonicalRegId);
+ } else {
+ String error = result.getErrorCodeName();
+ trackers.get(i).failed(error, error);
+ if (Constants.ERROR_NOT_REGISTERED.equals(error)
+ || Constants.ERROR_INVALID_REGISTRATION
+ .equals(error)) {
+ inactiveDevices.put(ids.get(i), new Date());
+ }
+ }
+ }
+ }
+ this.ids.clear();
+ this.trackers.clear();
+ }
+ }
+ public boolean isMock(Notifier notifier){
+ return notifier.getEnvironment() !=null ? notifier.getEnvironment().equals("mock") : false ;
+ }
+ public boolean delayRandom(Notifier notifier) {
+ boolean wasDelayed = false;
+ if (isMock(notifier)) {
+ try {
+ Thread.sleep(
+ new Random().nextInt(300)
+ );
+ wasDelayed = true;
+ } catch (InterruptedException ie) {
+ //delay was stopped
+ }
+ }
+ return wasDelayed;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/9d7901ae/stack/services/src/main/java/org/apache/usergrid/services/notifiers/NotifiersService.java
----------------------------------------------------------------------
diff --git a/stack/services/src/main/java/org/apache/usergrid/services/notifiers/NotifiersService.java b/stack/services/src/main/java/org/apache/usergrid/services/notifiers/NotifiersService.java
new file mode 100644
index 0000000..8d7964d
--- /dev/null
+++ b/stack/services/src/main/java/org/apache/usergrid/services/notifiers/NotifiersService.java
@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.usergrid.services.notifiers;
+
+import org.apache.usergrid.persistence.Notifier;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.apache.usergrid.services.*;
+import org.apache.usergrid.services.notifications.NotificationsService;
+import org.apache.usergrid.services.notifications.ProviderAdapter;
+
+import java.util.Arrays;
+import java.util.Set;
+
+public class NotifiersService extends AbstractCollectionService {
+
+ private static final Logger logger = LoggerFactory
+ .getLogger(NotifiersService.class);
+
+ public NotifiersService() {
+ super();
+ logger.info("/notifiers");
+ }
+
+ @Override
+ public ServiceResults postCollection(ServiceContext context)
+ throws Exception {
+
+ ServicePayload payload = context.getPayload();
+
+ NotificationsService ns = (NotificationsService) sm
+ .getService("notifications");
+ Set<String> providers = ns.getProviders();
+
+ String provider = payload.getStringProperty("provider");
+ if (!providers.contains(provider)) {
+ throw new IllegalArgumentException("provider must be one of: "
+ + Arrays.toString(providers.toArray()));
+ }
+
+ ProviderAdapter providerAdapter = ns.providerAdapters.get(provider);
+ providerAdapter.validateCreateNotifier(payload);
+
+ ServiceResults results = super.postCollection(context);
+
+ Notifier notifier = (Notifier) results.getEntity();
+ if (notifier != null) {
+ try {
+ ns.testConnection(notifier);
+ } catch (Exception e) {
+ logger.info("notifier testConnection() failed", e);
+ em.delete(notifier);
+ throw e;
+ }
+ }
+
+ return results;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/9d7901ae/stack/services/src/main/java/org/apache/usergrid/services/users/devices/notifications/NotificationsService.java
----------------------------------------------------------------------
diff --git a/stack/services/src/main/java/org/apache/usergrid/services/users/devices/notifications/NotificationsService.java b/stack/services/src/main/java/org/apache/usergrid/services/users/devices/notifications/NotificationsService.java
new file mode 100644
index 0000000..79972c6
--- /dev/null
+++ b/stack/services/src/main/java/org/apache/usergrid/services/users/devices/notifications/NotificationsService.java
@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.usergrid.services.users.devices.notifications;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class NotificationsService extends
+ org.apache.usergrid.services.notifications.NotificationsService {
+
+ private static final Logger logger = LoggerFactory
+ .getLogger(NotificationsService.class);
+
+ public NotificationsService() {
+ logger.info("/users/*/devices/*/notifications");
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/9d7901ae/stack/services/src/main/java/org/apache/usergrid/services/users/notifications/NotificationsService.java
----------------------------------------------------------------------
diff --git a/stack/services/src/main/java/org/apache/usergrid/services/users/notifications/NotificationsService.java b/stack/services/src/main/java/org/apache/usergrid/services/users/notifications/NotificationsService.java
new file mode 100644
index 0000000..4b536bc
--- /dev/null
+++ b/stack/services/src/main/java/org/apache/usergrid/services/users/notifications/NotificationsService.java
@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.usergrid.services.users.notifications;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class NotificationsService extends
+ org.apache.usergrid.services.notifications.NotificationsService {
+
+ private static final Logger logger = LoggerFactory
+ .getLogger(NotificationsService.class);
+
+ public NotificationsService() {
+ logger.info("/users/*/notifications");
+ }
+
+}
[2/8] git commit: add vcm
Posted by sf...@apache.org.
add vcm
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/9d2121c7
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/9d2121c7
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/9d2121c7
Branch: refs/heads/two-dot-o-push-notifications
Commit: 9d2121c7afba8a500289d478d83c762036617044
Parents: d2280df
Author: Shawn Feldman <sf...@apache.org>
Authored: Mon Aug 18 12:49:36 2014 -0600
Committer: Shawn Feldman <sf...@apache.org>
Committed: Mon Aug 18 12:49:36 2014 -0600
----------------------------------------------------------------------
stack/pom.xml | 10 ++++++++--
1 file changed, 8 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/9d2121c7/stack/pom.xml
----------------------------------------------------------------------
diff --git a/stack/pom.xml b/stack/pom.xml
index 35c3979..adc02fd 100644
--- a/stack/pom.xml
+++ b/stack/pom.xml
@@ -1405,7 +1405,7 @@
<artifactId>elasticsearch</artifactId>
<version>${elasticsearch.version}</version>
</dependency>
-
+
<dependency>
<groupId>com.relayrides</groupId>
<artifactId>pushy</artifactId>
@@ -1413,7 +1413,13 @@
<version>0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b</version>
</dependency>
- </dependencies>
+ <dependency>
+ <groupId>com.ganyo</groupId>
+ <artifactId>gcm-server</artifactId>
+ <version>1.0.2</version>
+ </dependency>
+
+ </dependencies>
</dependencyManagement>
[7/8] git commit: adding notificationsservice
Posted by sf...@apache.org.
adding notificationsservice
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/9d7901ae
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/9d7901ae
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/9d7901ae
Branch: refs/heads/two-dot-o-push-notifications
Commit: 9d7901ae388d80350439221f1a253a0846db7c71
Parents: 0a61101
Author: Shawn Feldman <sf...@apache.org>
Authored: Mon Aug 18 14:53:51 2014 -0600
Committer: Shawn Feldman <sf...@apache.org>
Committed: Mon Aug 18 14:53:51 2014 -0600
----------------------------------------------------------------------
.../index/exceptions/NoIndexException.java | 17 +-
.../index/query/tree/NotOperand.java | 17 +-
.../index/query/tree/WithinOperand.java | 17 +-
.../query/tree/StringLiteralTest.java | 17 +-
stack/services/pom.xml | 11 +
.../notifications/NotificationsService.java | 32 +
.../notifications/NotificationsService.java | 32 +
.../notifications/NotificationsService.java | 32 +
.../notifications/ConnectionException.java | 25 +
.../services/notifications/JobScheduler.java | 92 +++
.../notifications/NotificationBatchJob.java | 125 ++++
.../services/notifications/NotificationJob.java | 129 ++++
.../notifications/NotificationServiceProxy.java | 32 +
.../NotificationsQueueManager.java | 654 +++++++++++++++++++
.../notifications/NotificationsService.java | 327 ++++++++++
.../services/notifications/ProviderAdapter.java | 55 ++
.../services/notifications/QueueJob.java | 127 ++++
.../services/notifications/TaskManager.java | 196 ++++++
.../services/notifications/TaskTracker.java | 50 ++
.../services/notifications/TestAdapter.java | 106 +++
.../notifications/apns/APNsAdapter.java | 280 ++++++++
.../notifications/apns/APNsNotification.java | 101 +++
.../apns/FailedConnectionListener.java | 88 +++
.../apns/RejectedAPNsListener.java | 53 ++
.../notifications/apns/TestAPNsListener.java | 100 +++
.../apns/TestAPNsNotification.java | 124 ++++
.../services/notifications/gcm/GCMAdapter.java | 221 +++++++
.../services/notifiers/NotifiersService.java | 73 +++
.../notifications/NotificationsService.java | 32 +
.../notifications/NotificationsService.java | 32 +
30 files changed, 3165 insertions(+), 32 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/9d7901ae/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/exceptions/NoIndexException.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/exceptions/NoIndexException.java b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/exceptions/NoIndexException.java
index c505721..3043768 100644
--- a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/exceptions/NoIndexException.java
+++ b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/exceptions/NoIndexException.java
@@ -1,18 +1,19 @@
-/*******************************************************************************
- * Copyright 2012 Apigee Corporation
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
*
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
+ * http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
- ******************************************************************************/
+ */
package org.apache.usergrid.persistence.index.exceptions;
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/9d7901ae/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/query/tree/NotOperand.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/query/tree/NotOperand.java b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/query/tree/NotOperand.java
index 8588d24..35b41f3 100644
--- a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/query/tree/NotOperand.java
+++ b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/query/tree/NotOperand.java
@@ -1,18 +1,19 @@
-/*******************************************************************************
- * Copyright 2012 Apigee Corporation
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
*
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
+ * http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
- ******************************************************************************/
+ */
package org.apache.usergrid.persistence.index.query.tree;
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/9d7901ae/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/query/tree/WithinOperand.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/query/tree/WithinOperand.java b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/query/tree/WithinOperand.java
index 602fc34..9d913e3 100644
--- a/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/query/tree/WithinOperand.java
+++ b/stack/corepersistence/queryindex/src/main/java/org/apache/usergrid/persistence/index/query/tree/WithinOperand.java
@@ -1,18 +1,19 @@
-/*******************************************************************************
- * Copyright 2012 Apigee Corporation
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
*
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
+ * http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
- ******************************************************************************/
+ */
package org.apache.usergrid.persistence.index.query.tree;
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/9d7901ae/stack/corepersistence/queryindex/src/test/java/org/apache/usergrid/persistence/query/tree/StringLiteralTest.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/queryindex/src/test/java/org/apache/usergrid/persistence/query/tree/StringLiteralTest.java b/stack/corepersistence/queryindex/src/test/java/org/apache/usergrid/persistence/query/tree/StringLiteralTest.java
index f67cb11..161430a 100644
--- a/stack/corepersistence/queryindex/src/test/java/org/apache/usergrid/persistence/query/tree/StringLiteralTest.java
+++ b/stack/corepersistence/queryindex/src/test/java/org/apache/usergrid/persistence/query/tree/StringLiteralTest.java
@@ -1,18 +1,19 @@
-/*******************************************************************************
- * Copyright 2012 Apigee Corporation
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
*
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
+ * http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
- ******************************************************************************/
+ */
package org.apache.usergrid.persistence.query.tree;
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/9d7901ae/stack/services/pom.xml
----------------------------------------------------------------------
diff --git a/stack/services/pom.xml b/stack/services/pom.xml
index 3d81558..ace3c8d 100644
--- a/stack/services/pom.xml
+++ b/stack/services/pom.xml
@@ -549,6 +549,17 @@
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>com.relayrides</groupId>
+ <artifactId>pushy</artifactId>
+ <!-- The sha in the version is the git commit used in this build. Check out the pushy source, then this commit to build the library locally -->
+ <version>0.4-apigee-63dec68314c97c6e3ef40c88590e1b196d3ec55b</version>
+ </dependency>
+ <dependency>
+ <groupId>com.ganyo</groupId>
+ <artifactId>gcm-server</artifactId>
+ <version>1.0.2</version>
+ </dependency>
</dependencies>
</project>
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/9d7901ae/stack/services/src/main/java/org/apache/usergrid/services/devices/notifications/NotificationsService.java
----------------------------------------------------------------------
diff --git a/stack/services/src/main/java/org/apache/usergrid/services/devices/notifications/NotificationsService.java b/stack/services/src/main/java/org/apache/usergrid/services/devices/notifications/NotificationsService.java
new file mode 100644
index 0000000..c766c72
--- /dev/null
+++ b/stack/services/src/main/java/org/apache/usergrid/services/devices/notifications/NotificationsService.java
@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.usergrid.services.devices.notifications;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class NotificationsService extends
+ org.apache.usergrid.services.notifications.NotificationsService {
+
+ private static final Logger logger = LoggerFactory
+ .getLogger(NotificationsService.class);
+
+ public NotificationsService() {
+ logger.info("/devices/*/notifications");
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/9d7901ae/stack/services/src/main/java/org/apache/usergrid/services/groups/notifications/NotificationsService.java
----------------------------------------------------------------------
diff --git a/stack/services/src/main/java/org/apache/usergrid/services/groups/notifications/NotificationsService.java b/stack/services/src/main/java/org/apache/usergrid/services/groups/notifications/NotificationsService.java
new file mode 100644
index 0000000..9e965b5
--- /dev/null
+++ b/stack/services/src/main/java/org/apache/usergrid/services/groups/notifications/NotificationsService.java
@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.usergrid.services.groups.notifications;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class NotificationsService extends
+ org.apache.usergrid.services.notifications.NotificationsService {
+
+ private static final Logger logger = LoggerFactory
+ .getLogger(NotificationsService.class);
+
+ public NotificationsService() {
+ logger.info("/groups/*/notifications");
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/9d7901ae/stack/services/src/main/java/org/apache/usergrid/services/groups/users/devices/notifications/NotificationsService.java
----------------------------------------------------------------------
diff --git a/stack/services/src/main/java/org/apache/usergrid/services/groups/users/devices/notifications/NotificationsService.java b/stack/services/src/main/java/org/apache/usergrid/services/groups/users/devices/notifications/NotificationsService.java
new file mode 100644
index 0000000..7587bd0
--- /dev/null
+++ b/stack/services/src/main/java/org/apache/usergrid/services/groups/users/devices/notifications/NotificationsService.java
@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.usergrid.services.groups.users.devices.notifications;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class NotificationsService extends
+ org.apache.usergrid.services.notifications.NotificationsService {
+
+ private static final Logger logger = LoggerFactory
+ .getLogger(NotificationsService.class);
+
+ public NotificationsService() {
+ logger.info("/groups/*/users/*/devices/*/notifications");
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/9d7901ae/stack/services/src/main/java/org/apache/usergrid/services/notifications/ConnectionException.java
----------------------------------------------------------------------
diff --git a/stack/services/src/main/java/org/apache/usergrid/services/notifications/ConnectionException.java b/stack/services/src/main/java/org/apache/usergrid/services/notifications/ConnectionException.java
new file mode 100644
index 0000000..314834b
--- /dev/null
+++ b/stack/services/src/main/java/org/apache/usergrid/services/notifications/ConnectionException.java
@@ -0,0 +1,25 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.usergrid.services.notifications;
+
+import org.springframework.core.NestedCheckedException;
+
+public class ConnectionException extends NestedCheckedException {
+ public ConnectionException(String msg, Throwable cause) {
+ super(msg, cause);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/9d7901ae/stack/services/src/main/java/org/apache/usergrid/services/notifications/JobScheduler.java
----------------------------------------------------------------------
diff --git a/stack/services/src/main/java/org/apache/usergrid/services/notifications/JobScheduler.java b/stack/services/src/main/java/org/apache/usergrid/services/notifications/JobScheduler.java
new file mode 100644
index 0000000..c366a86
--- /dev/null
+++ b/stack/services/src/main/java/org/apache/usergrid/services/notifications/JobScheduler.java
@@ -0,0 +1,92 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.usergrid.services.notifications;
+
+import org.apache.usergrid.batch.service.SchedulerService;
+import org.apache.usergrid.persistence.EntityManager;
+import org.apache.usergrid.persistence.Notification;
+import org.apache.usergrid.persistence.entities.JobData;
+import org.apache.usergrid.services.ServiceManager;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class JobScheduler{
+ public static final long SCHEDULER_GRACE_PERIOD = 250;
+ private final EntityManager em;
+
+ protected ServiceManager sm;
+ private final Logger LOG = LoggerFactory.getLogger(NotificationsService.class);
+
+ public JobScheduler(ServiceManager sm,EntityManager em){
+ this.sm=sm; this.em = em;
+ }
+ public void scheduleBatchJob(Notification notification, long delay) throws Exception {
+
+ JobData jobData = new JobData();
+ jobData.setProperty("applicationId", sm.getApplicationId());
+ jobData.setProperty("notificationId", notification.getUuid());
+ jobData.setProperty("deliver", notification.getDeliver());
+
+ long soonestPossible = System.currentTimeMillis() + SCHEDULER_GRACE_PERIOD + delay;
+
+ SchedulerService scheduler = getSchedulerService();
+ scheduler.createJob("notificationBatchJob", soonestPossible, jobData);
+
+ LOG.info("notification {} batch scheduled for delivery", notification.getUuid());
+ }
+ public boolean scheduleQueueJob(Notification notification) throws Exception {
+ return scheduleQueueJob(notification,false);
+ }
+
+ public boolean scheduleQueueJob(Notification notification, boolean forceSchedule) throws Exception {
+
+ boolean scheduleInFuture = notification.getDeliver() != null;
+ long scheduleAt = (notification.getDeliver() != null) ? notification.getDeliver() : 0;
+ long soonestPossible = System.currentTimeMillis() + SCHEDULER_GRACE_PERIOD;
+ if (scheduleAt < soonestPossible) {
+ scheduleAt = soonestPossible;
+ scheduleInFuture = false;
+ }
+
+ // update Notification properties
+ if (notification.getStarted() == null || notification.getStarted() == 0) {
+ notification.setStarted(System.currentTimeMillis());
+ Map<String, Object> properties = new HashMap<String, Object>(2);
+ properties.put("started", notification.getStarted());
+ properties.put("state", notification.getState());
+ em.updateProperties(notification, properties);
+ }
+ boolean scheduled = scheduleInFuture || forceSchedule;
+ if(scheduled) {
+ JobData jobData = new JobData();
+ jobData.setProperty("applicationId", sm.getApplicationId());
+ jobData.setProperty("notificationId", notification.getUuid());
+ jobData.setProperty("deliver", notification.getDeliver());
+ SchedulerService scheduler = getSchedulerService();
+ scheduler.createJob("queueJob", scheduleAt, jobData);
+ }
+ LOG.info("notification {} scheduled for queuing", notification.getUuid());
+ return scheduled;
+ }
+ private SchedulerService getSchedulerService() {
+ return sm.getSchedulerService();
+ }
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/9d7901ae/stack/services/src/main/java/org/apache/usergrid/services/notifications/NotificationBatchJob.java
----------------------------------------------------------------------
diff --git a/stack/services/src/main/java/org/apache/usergrid/services/notifications/NotificationBatchJob.java b/stack/services/src/main/java/org/apache/usergrid/services/notifications/NotificationBatchJob.java
new file mode 100644
index 0000000..81006ce
--- /dev/null
+++ b/stack/services/src/main/java/org/apache/usergrid/services/notifications/NotificationBatchJob.java
@@ -0,0 +1,125 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.usergrid.services.notifications;
+
+
+import java.util.UUID;
+
+import javax.annotation.PostConstruct;
+
+import org.apache.usergrid.persistence.Notification;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import org.apache.usergrid.batch.Job;
+import org.apache.usergrid.batch.JobExecution;
+import org.apache.usergrid.metrics.MetricsFactory;
+import org.apache.usergrid.persistence.EntityManager;
+import org.apache.usergrid.persistence.EntityManagerFactory;
+import org.apache.usergrid.persistence.entities.JobData;
+import org.apache.usergrid.services.ServiceManager;
+import org.apache.usergrid.services.ServiceManagerFactory;
+
+import com.codahale.metrics.Histogram;
+import com.codahale.metrics.Meter;
+import com.codahale.metrics.Timer;
+
+
+@Component( "notificationBatchJob" )
+public class NotificationBatchJob implements Job {
+
+ private static final Logger logger = LoggerFactory.getLogger( NotificationBatchJob.class );
+ private Meter postSendMeter;
+ private Timer timerMetric;
+ private Meter requestMeter;
+ private Histogram cycleMetric;
+
+ @Autowired
+ private MetricsFactory metricsService;
+
+ @Autowired
+ private ServiceManagerFactory smf;
+
+ @Autowired
+ private EntityManagerFactory emf;
+
+
+ public NotificationBatchJob() {
+
+ }
+
+
+ @PostConstruct
+ void init() {
+ postSendMeter = metricsService.getMeter( NotificationsService.class, "post-send" );
+ requestMeter = metricsService.getMeter( NotificationBatchJob.class, "requests" );
+ timerMetric = metricsService.getTimer( NotificationBatchJob.class, "execution" );
+ cycleMetric = metricsService.getHistogram( NotificationsService.class, "cycle" );
+ }
+
+
+ public void execute( JobExecution jobExecution ) throws Exception {
+
+ Timer.Context timer = timerMetric.time();
+ requestMeter.mark();
+ logger.info( "execute NotificationBatchJob {}", jobExecution );
+
+ JobData jobData = jobExecution.getJobData();
+ UUID applicationId = ( UUID ) jobData.getProperty( "applicationId" );
+ ServiceManager sm = smf.getServiceManager( applicationId );
+ NotificationsService notificationsService = ( NotificationsService ) sm.getService( "notifications" );
+
+ EntityManager em = emf.getEntityManager( applicationId );
+
+ try {
+ if ( em == null ) {
+ logger.info( "no EntityManager for applicationId {}", applicationId );
+ return;
+ }
+
+ UUID notificationId = ( UUID ) jobData.getProperty( "notificationId" );
+ Notification notification = em.get( notificationId, Notification.class );
+ if ( notification == null ) {
+ logger.info( "notificationId {} no longer exists", notificationId );
+ return;
+ }
+
+
+ try {
+ notificationsService.getQueueManager().processBatchAndReschedule( notification, jobExecution );
+ }
+ catch ( Exception e ) {
+ logger.error( "execute NotificationBatchJob failed", e );
+ em.setProperty( notification, "errorMessage", e.getMessage() );
+ throw e;
+ }
+ finally {
+ long diff = System.currentTimeMillis() - notification.getCreated();
+ cycleMetric.update( diff );
+ postSendMeter.mark();
+ }
+ }
+ finally {
+ timer.stop();
+ }
+
+ logger.info( "execute NotificationBatch completed normally" );
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/9d7901ae/stack/services/src/main/java/org/apache/usergrid/services/notifications/NotificationJob.java
----------------------------------------------------------------------
diff --git a/stack/services/src/main/java/org/apache/usergrid/services/notifications/NotificationJob.java b/stack/services/src/main/java/org/apache/usergrid/services/notifications/NotificationJob.java
new file mode 100644
index 0000000..f80ee1e
--- /dev/null
+++ b/stack/services/src/main/java/org/apache/usergrid/services/notifications/NotificationJob.java
@@ -0,0 +1,129 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.usergrid.services.notifications;
+
+import java.util.UUID;
+
+import javax.annotation.PostConstruct;
+
+import org.apache.usergrid.persistence.Notification;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import org.apache.usergrid.batch.JobExecution;
+import org.apache.usergrid.batch.job.OnlyOnceJob;
+import org.apache.usergrid.metrics.MetricsFactory;
+import org.apache.usergrid.persistence.EntityManager;
+import org.apache.usergrid.persistence.EntityManagerFactory;
+import org.apache.usergrid.persistence.entities.JobData;
+import org.apache.usergrid.services.ServiceManager;
+import org.apache.usergrid.services.ServiceManagerFactory;
+
+import com.codahale.metrics.Histogram;
+import com.codahale.metrics.Meter;
+import com.codahale.metrics.Timer;
+
+
+@Component( "notificationJob" )
+public class NotificationJob extends OnlyOnceJob {
+
+
+ private static final Logger logger = LoggerFactory.getLogger( NotificationJob.class );
+
+ @Autowired
+ private MetricsFactory metricsService;
+
+ @Autowired
+ private ServiceManagerFactory smf;
+
+ @Autowired
+ private EntityManagerFactory emf;
+ private Meter requests;
+ private Timer execution;
+ private Histogram end;
+
+
+ public NotificationJob() {
+
+ }
+
+
+ @PostConstruct
+ void init() {
+ requests = metricsService.getMeter( NotificationJob.class, "requests" );
+ execution = metricsService.getTimer( NotificationJob.class, "execution" );
+ end = metricsService.getHistogram( QueueJob.class, "end" );
+ }
+
+
+ @Override
+ public void doJob( JobExecution jobExecution ) throws Exception {
+
+ Timer.Context timer = execution.time();
+ requests.mark();
+
+ logger.info( "execute NotificationJob {}", jobExecution );
+
+ JobData jobData = jobExecution.getJobData();
+ UUID applicationId = ( UUID ) jobData.getProperty( "applicationId" );
+ ServiceManager sm = smf.getServiceManager( applicationId );
+ NotificationsService notificationsService = ( NotificationsService ) sm.getService( "notifications" );
+
+ EntityManager em = emf.getEntityManager( applicationId );
+ try {
+ if ( em == null ) {
+ logger.info( "no EntityManager for applicationId {}", applicationId );
+ return;
+ }
+ UUID notificationId = ( UUID ) jobData.getProperty( "notificationId" );
+ Notification notification = em.get( notificationId, Notification.class );
+ if ( notification == null ) {
+ logger.info( "notificationId {} no longer exists", notificationId );
+ return;
+ }
+
+ try {
+ notificationsService.getQueueManager().processBatchAndReschedule( notification, jobExecution );
+ }
+ catch ( Exception e ) {
+ logger.error( "execute NotificationJob failed", e );
+ em.setProperty( notification, "errorMessage", e.getMessage() );
+ throw e;
+ }
+ finally {
+ long diff = System.currentTimeMillis() - notification.getCreated();
+ end.update( diff );
+ }
+ }
+ finally {
+ timer.stop();
+ }
+
+ logger.info( "execute NotificationJob completed normally" );
+ }
+
+
+ @Override
+ protected long getDelay( JobExecution execution ) throws Exception {
+ return TaskManager.BATCH_DEATH_PERIOD;
+ }
+
+
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/9d7901ae/stack/services/src/main/java/org/apache/usergrid/services/notifications/NotificationServiceProxy.java
----------------------------------------------------------------------
diff --git a/stack/services/src/main/java/org/apache/usergrid/services/notifications/NotificationServiceProxy.java b/stack/services/src/main/java/org/apache/usergrid/services/notifications/NotificationServiceProxy.java
new file mode 100644
index 0000000..908482f
--- /dev/null
+++ b/stack/services/src/main/java/org/apache/usergrid/services/notifications/NotificationServiceProxy.java
@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.usergrid.services.notifications;
+
+import org.apache.usergrid.persistence.Notification;
+import org.apache.usergrid.persistence.Notifier;
+
+import java.util.Set;
+
+/**
+ * Created by ApigeeCorporation on 8/6/14.
+ */
+public interface NotificationServiceProxy {
+
+ public void finishedBatch(Notification notification, long successes, long failures) throws Exception;
+
+ void asyncCheckForInactiveDevices(Set<Notifier> notifiers) throws Exception;
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/9d7901ae/stack/services/src/main/java/org/apache/usergrid/services/notifications/NotificationsQueueManager.java
----------------------------------------------------------------------
diff --git a/stack/services/src/main/java/org/apache/usergrid/services/notifications/NotificationsQueueManager.java b/stack/services/src/main/java/org/apache/usergrid/services/notifications/NotificationsQueueManager.java
new file mode 100644
index 0000000..95dfb8e
--- /dev/null
+++ b/stack/services/src/main/java/org/apache/usergrid/services/notifications/NotificationsQueueManager.java
@@ -0,0 +1,654 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.usergrid.services.notifications;
+
+import com.codahale.metrics.Counter;
+import com.codahale.metrics.Histogram;
+import com.codahale.metrics.Meter;
+import com.google.common.cache.*;
+import org.apache.usergrid.batch.JobExecution;
+import org.apache.usergrid.metrics.MetricsFactory;
+import org.apache.usergrid.mq.Message;
+import org.apache.usergrid.mq.QueueManager;
+import org.apache.usergrid.mq.QueueQuery;
+import org.apache.usergrid.mq.QueueResults;
+import org.apache.usergrid.persistence.*;
+import org.apache.usergrid.persistence.entities.*;
+import org.apache.usergrid.persistence.index.query.Query;
+import org.apache.usergrid.services.notifications.apns.APNsAdapter;
+import org.apache.usergrid.services.notifications.gcm.GCMAdapter;
+import org.apache.usergrid.utils.InflectionUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import rx.Observable;
+import rx.Subscriber;
+import rx.functions.Func1;
+import rx.schedulers.Schedulers;
+
+import java.util.*;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * Created by ApigeeCorporation on 8/13/14.
+ */
+public class NotificationsQueueManager implements NotificationServiceProxy {
+ private static final String NOTIFICATION_CONCURRENT_BATCHES = "notification.concurrent.batches";
+ private static final long CONSECUTIVE_EMPTY_QUEUES = 10;
+
+ private static final Logger LOG = LoggerFactory.getLogger(NotificationsQueueManager.class);
+
+ //this is for tests, will not mark initial post complete, set to false for tests
+ public static boolean IS_TEST = false;
+ private final Meter sendMeter;
+ private final Histogram queueSize;
+ private final Counter outstandingQueue;
+ private static ExecutorService INACTIVE_DEVICE_CHECK_POOL = Executors.newFixedThreadPool(5);
+ public static final String NOTIFIER_ID_POSTFIX = ".notifier.id";
+ public static int BATCH_SIZE = 1000;
+ // timeout for message queue transaction
+ public static final long MESSAGE_TRANSACTION_TIMEOUT = TaskManager.MESSAGE_TRANSACTION_TIMEOUT;
+ // If this property is set Notifications are automatically expired in
+ // the isOkToSent() method after the specified number of milliseconds
+ public static final String PUSH_AUTO_EXPIRE_AFTER_PROPNAME = "usergrid.push-auto-expire-after";
+ private final EntityManager em;
+ private final QueueManager qm;
+ private final JobScheduler jobScheduler;
+ private final Properties props;
+ private final InflectionUtils utils;
+ private AtomicLong consecutiveEmptyQueues = new AtomicLong();
+ private Long pushAutoExpireAfter = null;
+ static final String MESSAGE_PROPERTY_DEVICE_UUID = "deviceUUID";
+
+ public final Map<String, ProviderAdapter> providerAdapters = new HashMap<String, ProviderAdapter>(3);
+ {
+ providerAdapters.put("apple", APNS_ADAPTER);
+ providerAdapters.put("google", new GCMAdapter());
+ providerAdapters.put("noop", TEST_ADAPTER);
+ };
+ // these 2 can be static, but GCM can't. future: would be nice to get gcm
+ // static as well...
+ public static ProviderAdapter APNS_ADAPTER = new APNsAdapter();
+ public static ProviderAdapter TEST_ADAPTER = new TestAdapter();
+
+ //cache to retrieve push manager, cached per notifier, so many notifications will get same push manager
+ private static LoadingCache<EntityManager, HashMap<Object,Notifier>> notifierCacheMap = CacheBuilder
+ .newBuilder()
+ .expireAfterWrite(5, TimeUnit.MINUTES)
+ .build(new CacheLoader<EntityManager, HashMap<Object, Notifier>>() {
+ @Override
+ public HashMap<Object, Notifier> load(EntityManager em) {
+ HashMap<Object, Notifier> notifierHashMap = new HashMap<Object, Notifier>();
+ Query query = new Query();
+ query.setCollection("notifiers");
+ query.setLimit(50);
+ PathQuery<Notifier> pathQuery = new PathQuery<Notifier>(em.getApplicationRef(), query);
+ Iterator<Notifier> notifierIterator = pathQuery.iterator(em);
+ while (notifierIterator.hasNext()) {
+ Notifier notifier = notifierIterator.next();
+ notifierHashMap.put(notifier.getName().toLowerCase(), notifier);
+ notifierHashMap.put(notifier.getUuid(), notifier);
+ notifierHashMap.put(notifier.getUuid().toString(), notifier);
+
+ }
+ return notifierHashMap;
+ }
+ });;
+
+ public NotificationsQueueManager(JobScheduler jobScheduler, EntityManager entityManager, Properties props, QueueManager queueManager, MetricsFactory metricsFactory){
+ this.em = entityManager;
+ this.qm = queueManager;
+ this.jobScheduler = jobScheduler;
+ this.props = props;
+ this.sendMeter = metricsFactory.getMeter(NotificationsService.class, "send");
+ this.queueSize = metricsFactory.getHistogram(NotificationsService.class, "queue_size");
+ this.outstandingQueue = metricsFactory.getCounter(NotificationsService.class,"current_queue");
+ utils = new InflectionUtils();
+ }
+
+ public boolean scheduleQueueJob(Notification notification) throws Exception{
+ return jobScheduler.scheduleQueueJob(notification);
+ }
+
+ public void queueNotification(final Notification notification, final JobExecution jobExecution) throws Exception {
+ if (notification.getCanceled() == Boolean.TRUE) {
+ LOG.info("notification " + notification.getUuid() + " canceled");
+ if (jobExecution != null) {
+ jobExecution.killed();
+ }
+ return;
+ }
+
+ long startTime = System.currentTimeMillis();
+ LOG.info("notification {} start queuing", notification.getUuid());
+ final PathQuery<Device> pathQuery = notification.getPathQuery(); //devices query
+ final AtomicInteger deviceCount = new AtomicInteger(); //count devices so you can make a judgement on batching
+ final AtomicInteger batchCount = new AtomicInteger(); //count devices so you can make a judgement on batching
+ final int numCurrentBatchesConfig = getNumConcurrentBatches();
+
+ final ConcurrentLinkedQueue<String> errorMessages = new ConcurrentLinkedQueue<String>(); //build up list of issues
+
+ //get devices in querystring, and make sure you have access
+ if (pathQuery != null) {
+ final Iterator<Device> iterator = pathQuery.iterator(em);
+ //if there are more pages (defined by PAGE_SIZE) you probably want this to be async, also if this is already a job then don't reschedule
+ if (iterator instanceof ResultsIterator && ((ResultsIterator) iterator).hasPages() && jobExecution == null) {
+ jobScheduler.scheduleQueueJob(notification, true);
+ return;
+ }
+
+ Observable.create(new IteratorObservable<Entity>(iterator)).parallel(new Func1<Observable<Entity>, Observable<Entity>>() {
+ @Override
+ public Observable<Entity> call(Observable<Entity> deviceObservable) {
+ return deviceObservable.map( new Func1<Entity,Entity>() {
+ @Override
+ public Entity call(Entity entity) {
+ try {
+ List<EntityRef> devicesRef = getDevices(entity); // resolve group
+ boolean maySchedule = false;
+ for (EntityRef deviceRef : devicesRef) {
+ maySchedule |= deviceCount.incrementAndGet() % BATCH_SIZE == 0;
+ Message message = new Message();
+ message.setProperty(MESSAGE_PROPERTY_DEVICE_UUID, deviceRef.getUuid());
+ qm.postToQueue( getJobQueueName(notification), message);
+ if(notification.getQueued() == null){
+ // update queued time
+ notification.setQueued(System.currentTimeMillis());
+ em.update(notification);
+ }
+ }
+
+ //start working if you are on a large batch,
+ if (maySchedule && numCurrentBatchesConfig >= batchCount.incrementAndGet()) {
+ processBatchAndReschedule(notification, jobExecution);
+ }
+
+ if(devicesRef.size() <=0){
+ errorMessages.add("Could not find devices for entity: "+entity.getUuid());
+ }
+
+ } catch (Exception deviceLoopException) {
+ LOG.error("Failed to add devices", deviceLoopException);
+ errorMessages.add("Failed to add devices for entity: "+entity.getUuid() + " error:"+ deviceLoopException);
+ }
+ return entity;
+ }
+ });
+ }
+ }, Schedulers.io()).toBlocking().lastOrDefault(null);
+
+ }
+
+ batchCount.set(Math.min(numCurrentBatchesConfig, batchCount.get()));
+ // update queued time
+ Map<String, Object> properties = new HashMap<String, Object>(2);
+ properties.put("queued", notification.getQueued());
+ properties.put("state", notification.getState());
+
+ //do i have devices, and have i already started batching.
+ if (deviceCount.get()>0 ) {
+ if(batchCount.get() <= 0) {
+ processBatchAndReschedule(notification, jobExecution);
+ }
+ } else {
+ //if i'm in a test value will be false, do not mark finished for test orchestration, not ideal need real tests
+ if(!IS_TEST) {
+ finishedBatch(notification, 0, 0);
+ errorMessages.add("No devices for notification " + notification.getUuid());
+ }
+ }
+
+ if(!IS_TEST && errorMessages.size()>0){
+ properties.put("deliveryErrors", errorMessages.toArray());
+ if(notification.getErrorMessage()==null){
+ notification.setErrorMessage("There was a problem delivering all of your notifications. See deliveryErrors in properties");
+ }
+ }
+ notification.addProperties(properties);
+ em.update(notification);
+ long elapsed = notification.getQueued() - startTime;
+
+ if (LOG.isInfoEnabled()) {
+ StringBuilder sb = new StringBuilder();
+ sb.append("notification ").append(notification.getUuid());
+ sb.append(" done queuing to ").append(deviceCount);
+ sb.append(" devices in ").append(elapsed).append(" ms");
+ LOG.info(sb.toString());
+ }
+ }
+ public void finishedBatch(Notification notification, long successes, long failures) throws Exception {
+ finishedBatch(notification,successes,failures,false);
+ }
+ public void finishedBatch(Notification notification, long successes, long failures, boolean overrideComplete) throws Exception {
+ if (LOG.isDebugEnabled()) {
+ StringBuilder sb = new StringBuilder();
+ sb.append("finishedBatch ").append(notification.getUuid());
+ sb.append(" successes: ").append(successes);
+ sb.append(" failures: ").append(failures);
+ LOG.info(sb.toString());
+ }
+
+ notification = em.get(notification, Notification.class); // refresh data
+ notification.updateStatistics(successes, failures);
+ notification.setModified(System.currentTimeMillis());
+ Map<String, Object> properties = new HashMap<String, Object>(4);
+ properties.put("statistics", notification.getStatistics());
+ properties.put("modified", notification.getModified());
+
+ if (isNotificationDeliveryComplete(notification) || overrideComplete) {
+ notification.setFinished(notification.getModified());
+ properties.put("finished", notification.getModified());
+ properties.put("state", notification.getState());
+ long elapsed = notification.getFinished()
+ - notification.getStarted();
+ long sent = notification.getStatistics().get("sent");
+ long errors = notification.getStatistics().get("errors");
+
+ if (LOG.isInfoEnabled()) {
+ StringBuilder sb = new StringBuilder();
+ sb.append("notification ").append(notification.getUuid());
+ sb.append(" done sending to ").append(sent + errors);
+ sb.append(" devices in ").append(elapsed).append(" ms");
+ LOG.info(sb.toString());
+ }
+
+ } else {
+ LOG.info("notification finished batch: {}",
+ notification.getUuid());
+ }
+ em.updateProperties(notification, properties);
+
+ Set<Notifier> notifiers = new HashSet<Notifier>(getNotifierMap().values()); // remove dups
+ asyncCheckForInactiveDevices(notifiers);
+ }
+
+ private HashMap<Object,Notifier> getNotifierMap(){
+ try{
+ return notifierCacheMap.get(em);
+ }catch (ExecutionException ee){
+ LOG.error("failed to get from cache",ee);
+ return new HashMap<Object, Notifier>();
+ }
+ }
+ /*
+ * returns partial list of Device EntityRefs (up to BATCH_SIZE) - empty when
+ * done w/ delivery
+ */
+ private QueueResults getDeliveryBatch(EntityRef notification, int batchSize) throws Exception {
+
+ QueueQuery qq = new QueueQuery();
+ qq.setLimit(batchSize);
+ qq.setTimeout(MESSAGE_TRANSACTION_TIMEOUT);
+ QueueResults results = qm.getFromQueue(getJobQueueName(notification), qq);
+ LOG.debug("got batch of {} devices for notification {}",
+ results.size(), notification.getUuid());
+ return results;
+ }
+ /**
+ * executes a Notification batch and schedules the next one - called by
+ * NotificationBatchJob
+ */
+ public void processBatchAndReschedule(Notification notification, JobExecution jobExecution) throws Exception {
+ if (!isOkToSend(notification,jobExecution)) {
+ LOG.info("notification " + notification.getUuid() + " canceled");
+ if (jobExecution != null) {
+ jobExecution.killed();
+ }
+ return;
+ }
+
+ LOG.debug("processing batch {}", notification.getUuid());
+
+ QueueResults queueResults = getDeliveryBatch(notification, jobExecution == null ? BATCH_SIZE/2 : BATCH_SIZE ); //if first run then throttle the batch down by factor of 2 if its a job then try to grab alot of notifications and run them all
+
+ long reschedule_delay = jobScheduler.SCHEDULER_GRACE_PERIOD;
+ final TaskManager taskManager = new TaskManager( em, this, qm, notification, queueResults);
+ if (queueResults.size() > 0) {
+ consecutiveEmptyQueues.set(0);
+ sendBatchToProviders(taskManager, notification, queueResults.getMessages());
+ }else{
+ consecutiveEmptyQueues.incrementAndGet();
+ }
+
+ if (!isNotificationDeliveryComplete(notification) && consecutiveEmptyQueues.get() <= CONSECUTIVE_EMPTY_QUEUES) {
+ if(jobExecution==null) {
+ jobScheduler.scheduleBatchJob(notification, reschedule_delay);
+ }else{
+ processBatchAndReschedule(notification, jobExecution);
+ }
+ } else {
+ consecutiveEmptyQueues.set(0);
+ finishedBatch(notification, 0, 0,true);
+ }
+
+ LOG.debug("finished processing batch");
+ }
+
+ /**
+ * send batches of notifications to provider
+ * @param taskManager
+ * @param notification
+ * @param queueResults
+ * @throws Exception
+ */
+ private void sendBatchToProviders(final TaskManager taskManager, final Notification notification, List<Message> queueResults) throws Exception {
+
+ LOG.info("sending batch of {} devices for Notification: {}", queueResults.size(), notification.getUuid());
+ final Map<Object, Notifier> notifierMap = getNotifierMap();
+ queueSize.update(queueResults.size());
+ final Map<String, Object> payloads = notification.getPayloads();
+ final Map<String, Object> translatedPayloads = translatePayloads(payloads, notifierMap);
+ try {
+ Observable
+ .from(queueResults)
+ .parallel(new Func1<Observable<Message>, Observable<Message>>() {
+ @Override
+ public Observable<Message> call(Observable<Message> messageObservable) {
+ return messageObservable.map(new Func1<Message, Message>() {
+ @Override
+ public Message call(Message message) {
+ UUID deviceUUID = (UUID) message.getObjectProperty(MESSAGE_PROPERTY_DEVICE_UUID);
+ boolean foundNotifier = false;
+ for (Map.Entry<String, Object> entry : payloads.entrySet()) {
+ try {
+ String payloadKey = entry.getKey();
+ Notifier notifier = notifierMap.get(payloadKey.toLowerCase());
+ EntityRef deviceRef = new SimpleEntityRef(Device.ENTITY_TYPE, deviceUUID);
+
+ String providerId;
+ try {
+ providerId = getProviderId(deviceRef, notifier);
+ if (providerId == null) {
+ LOG.debug("Provider not found.{} {}", deviceRef,notifier.getName());
+ continue;
+ }
+ } catch (Exception providerException) {
+ LOG.error("Exception getting provider.", providerException);
+ continue;
+ }
+ Object payload = translatedPayloads.get(payloadKey);
+
+ Receipt receipt = new Receipt(notification.getUuid(), providerId, payload,deviceUUID);
+ TaskTracker tracker = new TaskTracker(notifier, taskManager, receipt, deviceUUID);
+ if (payload == null) {
+ LOG.debug("selected device {} for notification {} doesn't have a valid payload. skipping.", deviceUUID, notification.getUuid());
+ try {
+ tracker.failed(0, "failed to match payload to " + payloadKey + " notifier");
+ } catch (Exception e) {
+ LOG.debug("failed to mark device failed" + e);
+ }
+ continue;
+ }
+
+ if (LOG.isDebugEnabled()) {
+ StringBuilder sb = new StringBuilder();
+ sb.append("sending notification ").append(notification.getUuid());
+ sb.append(" to device ").append(deviceUUID);
+ LOG.debug(sb.toString());
+ }
+
+ try {
+ ProviderAdapter providerAdapter = providerAdapters.get(notifier.getProvider());
+ providerAdapter.sendNotification(providerId, notifier, payload, notification, tracker);
+
+ } catch (Exception e) {
+ try {
+ tracker.failed(0, e.getMessage());
+ } catch (Exception trackerException) {
+ LOG.error("tracker failed", trackerException);
+ }
+ }
+ foundNotifier = true;
+ } finally {
+ sendMeter.mark();
+ }
+ }
+ if (!foundNotifier) {
+ try {
+ taskManager.skip(deviceUUID);
+ } catch (Exception trackerException) {
+ LOG.error("failed on skip", trackerException);
+ }
+ }
+ return message;
+ }
+ });
+ }
+ }, Schedulers.io())
+ .toBlocking()
+ .lastOrDefault(null);
+
+ //for gcm this will actually send notification
+ for (ProviderAdapter providerAdapter : providerAdapters.values()) {
+ try {
+ providerAdapter.doneSendingNotifications();
+ } catch (Exception e) {
+ LOG.error("providerAdapter.doneSendingNotifications: ", e);
+ }
+ }
+
+ } finally {
+ outstandingQueue.dec();
+ LOG.info("finished sending batch for notification {}", notification.getUuid());
+ }
+
+ }
+
+ private Map<String, Object> translatePayloads(Map<String, Object> payloads, Map<Object, Notifier> notifierMap) throws Exception {
+ Map<String, Object> translatedPayloads = new HashMap<String, Object>(
+ payloads.size());
+ for (Map.Entry<String, Object> entry : payloads.entrySet()) {
+ String payloadKey = entry.getKey();
+ Object payloadValue = entry.getValue();
+ Notifier notifier = notifierMap.get(payloadKey);
+ if (notifier != null) {
+ ProviderAdapter providerAdapter = providerAdapters.get(notifier
+ .getProvider());
+ if (providerAdapter != null) {
+ Object translatedPayload = payloadValue != null ? providerAdapter
+ .translatePayload(payloadValue) : null;
+ if (translatedPayload != null) {
+ translatedPayloads.put(payloadKey, translatedPayload);
+ }
+ }
+ }
+ }
+ return translatedPayloads;
+ }
+
+ private int getNumConcurrentBatches() {
+ return Integer.parseInt(props.getProperty(NOTIFICATION_CONCURRENT_BATCHES, "1"));
+ }
+
+ private static final class IteratorObservable<T> implements Observable.OnSubscribe<T> {
+ private final Iterator<T> input;
+ private IteratorObservable( final Iterator input ) {this.input = input;}
+
+ @Override
+ public void call( final Subscriber<? super T> subscriber ) {
+
+ /**
+ * You would replace this code with your file reading. Instead of emitting from an iterator,
+ * you would create a bean object that represents the entity, and then emit it
+ */
+
+ try {
+ while ( !subscriber.isUnsubscribed() && input.hasNext() ) {
+ //send our input to the next
+ subscriber.onNext( (T) input.next() );
+ }
+
+ //tell the subscriber we don't have any more data
+ subscriber.onCompleted();
+ }
+ catch ( Throwable t ) {
+ LOG.error("failed on subscriber",t);
+ subscriber.onError( t );
+ }
+ }
+ }
+
+ public void asyncCheckForInactiveDevices(Set<Notifier> notifiers) throws Exception {
+ for (final Notifier notifier : notifiers) {
+ INACTIVE_DEVICE_CHECK_POOL.execute(new Runnable() {
+ @Override
+ public void run() {
+ try {
+ checkForInactiveDevices(notifier);
+ } catch (Exception e) {
+ LOG.error("checkForInactiveDevices", e); // not
+ // essential so
+ // don't fail,
+ // but log
+ }
+ }
+ });
+ }
+ }
+
+ /** gets the list of inactive devices from the Provider and updates them */
+ private void checkForInactiveDevices(Notifier notifier) throws Exception {
+ ProviderAdapter providerAdapter = providerAdapters.get(notifier
+ .getProvider());
+ if (providerAdapter != null) {
+ LOG.debug("checking notifier {} for inactive devices", notifier);
+ Map<String, Date> inactiveDeviceMap = providerAdapter
+ .getInactiveDevices(notifier, em);
+
+ if (inactiveDeviceMap != null && inactiveDeviceMap.size() > 0) {
+ LOG.debug("processing {} inactive devices",
+ inactiveDeviceMap.size());
+ Map<String, Object> clearPushtokenMap = new HashMap<String, Object>(
+ 2);
+ clearPushtokenMap.put(notifier.getName() + NOTIFIER_ID_POSTFIX,
+ "");
+ clearPushtokenMap.put(notifier.getUuid() + NOTIFIER_ID_POSTFIX,
+ "");
+
+ // todo: this could be done in a single query
+ for (Map.Entry<String, Date> entry : inactiveDeviceMap
+ .entrySet()) {
+ // name
+ Query query = new Query();
+ query.addEqualityFilter(notifier.getName()
+ + NOTIFIER_ID_POSTFIX, entry.getKey());
+ Results results = em.searchCollection(em.getApplication(),
+ "devices", query);
+ for (Entity e : results.getEntities()) {
+ em.updateProperties(e, clearPushtokenMap);
+ }
+ // uuid
+ query = new Query();
+ query.addEqualityFilter(notifier.getUuid()
+ + NOTIFIER_ID_POSTFIX, entry.getKey());
+ results = em.searchCollection(em.getApplication(),
+ "devices", query);
+ for (Entity e : results.getEntities()) {
+ em.updateProperties(e, clearPushtokenMap);
+ }
+ }
+ }
+ LOG.debug("finished checking notifier {} for inactive devices",
+ notifier);
+ }
+ }
+
+ private boolean isOkToSend(Notification notification,
+ JobExecution jobExecution) {
+ String autoExpireAfterString = props.getProperty(PUSH_AUTO_EXPIRE_AFTER_PROPNAME);
+
+ if (autoExpireAfterString != null) {
+ pushAutoExpireAfter = Long.parseLong(autoExpireAfterString);
+ }
+ if (pushAutoExpireAfter != null) {
+ if (notification.getCreated() < System.currentTimeMillis() - pushAutoExpireAfter) {
+ notification.setExpire(System.currentTimeMillis() - 1L);
+ }
+ }
+
+ if (notification.getFinished() != null) {
+ LOG.info("notification {} already processed. not sending.",
+ notification.getUuid());
+ return false;
+ }
+ if (notification.getCanceled() == Boolean.TRUE) {
+ LOG.info("notification {} canceled. not sending.",
+ notification.getUuid());
+ return false;
+ }
+ if (notification.isExpired()) {
+ LOG.info("notification {} expired. not sending.",
+ notification.getUuid());
+ return false;
+ }
+ if (jobExecution != null
+ && notification.getDeliver() != null
+ && !notification.getDeliver().equals(jobExecution.getJobData().getProperty("deliver"))
+ ) {
+ LOG.info("notification {} was rescheduled. not sending.",
+ notification.getUuid());
+ return false;
+ }
+
+ return true;
+ }
+
+ private List<EntityRef> getDevices(EntityRef ref) throws Exception {
+ List<EntityRef> devices = Collections.EMPTY_LIST;
+ if ("device".equals(ref.getType())) {
+ devices = Collections.singletonList(ref);
+ } else if ("user".equals(ref.getType())) {
+ devices = em.getCollection(ref, "devices", null, Query.MAX_LIMIT, Query.Level.REFS, false).getRefs();
+ } else if ("group".equals(ref.getType())) {
+ devices = new ArrayList<EntityRef>();
+ for (EntityRef r : em.getCollection(ref, "users", null, Query.MAX_LIMIT, Query.Level.REFS, false).getRefs()) {
+ devices.addAll(getDevices(r));
+ }
+ }
+ return devices;
+ }
+
+
+ private String getProviderId(EntityRef device, Notifier notifier) throws Exception {
+ try {
+ Object value = em.getProperty(device, notifier.getName() + NOTIFIER_ID_POSTFIX);
+ if (value == null) {
+ value = em.getProperty(device, notifier.getUuid() + NOTIFIER_ID_POSTFIX);
+ }
+ return value != null ? value.toString() : null;
+ } catch (Exception e) {
+ LOG.error("Errer getting provider ID, proceding with rest of batch", e);
+ return null;
+ }
+ }
+
+
+ private boolean isNotificationDeliveryComplete(Notification notification) throws Exception {
+ if(notification.getQueued() == null){
+ return false;
+ }
+ String queuePath = getJobQueueName(notification);
+ return !qm.hasPendingReads(queuePath, null)
+ && !qm.hasOutstandingTransactions(queuePath, null)
+ && !qm.hasMessagesInQueue(queuePath, null);
+ }
+
+ private String getJobQueueName(EntityRef entityRef) {
+ return utils.pluralize(entityRef.getType()) + "/" + entityRef.getUuid();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/9d7901ae/stack/services/src/main/java/org/apache/usergrid/services/notifications/NotificationsService.java
----------------------------------------------------------------------
diff --git a/stack/services/src/main/java/org/apache/usergrid/services/notifications/NotificationsService.java b/stack/services/src/main/java/org/apache/usergrid/services/notifications/NotificationsService.java
new file mode 100644
index 0000000..7dd6503
--- /dev/null
+++ b/stack/services/src/main/java/org/apache/usergrid/services/notifications/NotificationsService.java
@@ -0,0 +1,327 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.usergrid.services.notifications;
+
+import java.util.*;
+
+import com.codahale.metrics.*;
+import com.codahale.metrics.Timer;
+import org.apache.usergrid.metrics.MetricsFactory;
+import org.apache.usergrid.mq.Message;
+import org.apache.usergrid.persistence.*;
+import org.apache.usergrid.persistence.index.query.Identifier;
+import org.apache.usergrid.persistence.index.query.Query;
+import org.apache.usergrid.services.*;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.usergrid.persistence.entities.Device;
+import org.apache.usergrid.persistence.exceptions.RequiredPropertyNotFoundException;
+import org.apache.usergrid.services.exceptions.ForbiddenServiceOperationException;
+import static org.apache.usergrid.utils.InflectionUtils.pluralize;
+
+import org.apache.usergrid.services.notifications.apns.APNsAdapter;
+import org.apache.usergrid.services.notifications.gcm.GCMAdapter;
+import rx.Observable;
+import rx.functions.Func1;
+import rx.schedulers.Schedulers;
+
+public class NotificationsService extends AbstractCollectionService {
+
+
+ private MetricsFactory metricsService;
+ private Meter sendMeter;
+ private Meter postMeter;
+ private Timer postTimer;
+ private Histogram queueSize;
+ private Counter outstandingQueue;
+
+ private static final int PAGE = 100;
+ private static final Logger LOG = LoggerFactory.getLogger(NotificationsService.class);
+
+
+ public static final String NOTIFIER_ID_POSTFIX = ".notifier.id";
+
+ static final String MESSAGE_PROPERTY_DEVICE_UUID = "deviceUUID";
+
+ static {
+ Message.MESSAGE_PROPERTIES.put(
+ MESSAGE_PROPERTY_DEVICE_UUID, UUID.class);
+ }
+
+ // these 2 can be static, but GCM can't. future: would be nice to get gcm
+ // static as well...
+ public static ProviderAdapter APNS_ADAPTER = new APNsAdapter();
+ public static ProviderAdapter TEST_ADAPTER = new TestAdapter();
+
+ public final Map<String, ProviderAdapter> providerAdapters =
+ new HashMap<String, ProviderAdapter>(3);
+ {
+ providerAdapters.put("apple", APNS_ADAPTER);
+ providerAdapters.put("google", new GCMAdapter());
+ providerAdapters.put("noop", TEST_ADAPTER);
+ }
+
+ private NotificationsQueueManager notificationQueueManager;
+ private long gracePeriod;
+
+ public NotificationsService() {
+ LOG.info("/notifications");
+ }
+
+ @Override
+ public void init( ServiceInfo info ) {
+ super.init(info);
+ metricsService = getApplicationContext().getBean(MetricsFactory.class);
+ sendMeter = metricsService.getMeter(NotificationsService.class, "send");
+ postMeter = metricsService.getMeter(NotificationsService.class, "requests");
+ postTimer = metricsService.getTimer(this.getClass(), "execution_rest");
+ queueSize = metricsService.getHistogram(NotificationsService.class, "queue_size");
+ outstandingQueue = metricsService.getCounter(NotificationsService.class,"current_queue");
+ JobScheduler jobScheduler = new JobScheduler(sm,em);
+ notificationQueueManager = new NotificationsQueueManager(jobScheduler,em,sm.getProperties(),sm.getQueueManager(),metricsService);
+ gracePeriod = jobScheduler.SCHEDULER_GRACE_PERIOD;
+ }
+
+ public NotificationsQueueManager getQueueManager(){
+ return notificationQueueManager;
+ }
+
+ @Override
+ public ServiceContext getContext(ServiceAction action,
+ ServiceRequest request, ServiceResults previousResults,
+ ServicePayload payload) throws Exception {
+
+ ServiceContext context = super.getContext(action, request, previousResults, payload);
+
+ if (action == ServiceAction.POST) {
+ context.setQuery(null); // we don't use this, and it must be null to
+ // force the correct execution path
+ }
+ return context;
+ }
+
+ @Override
+ public ServiceResults postCollection(ServiceContext context) throws Exception {
+ Timer.Context timer = postTimer.time();
+ postMeter.mark();
+ try {
+ validate(null, context.getPayload());
+ PathQuery<Device> pathQuery = getPathQuery(context.getRequest().getOriginalParameters());
+ context.getProperties().put("state", Notification.State.CREATED);
+ context.getProperties().put("pathQuery", pathQuery);
+ context.setOwner(sm.getApplication());
+ ServiceResults results = super.postCollection(context);
+ Notification notification = (Notification) results.getEntity();
+ if(!notificationQueueManager.scheduleQueueJob(notification)){
+ notificationQueueManager.queueNotification(notification, null);
+ }
+ outstandingQueue.inc();
+ // future: somehow return 202?
+ return results;
+ }finally {
+ timer.stop();
+ }
+ }
+
+ private PathQuery<Device> getPathQuery(List<ServiceParameter> parameters)
+ throws Exception {
+
+ PathQuery pathQuery = null;
+ for (int i = 0; i < parameters.size() - 1; i += 2) {
+ String collection = pluralize(parameters.get(i).getName());
+ ServiceParameter sp = parameters.get(i + 1);
+ org.apache.usergrid.persistence.index.query.Query query = sp.getQuery();
+ if (query == null) {
+ query = new Query();
+ query.addIdentifier(sp.getIdentifier());
+ }
+ query.setLimit(PAGE);
+ query.setCollection(collection);
+
+ if (pathQuery == null) {
+ pathQuery = new PathQuery(em.getApplicationRef(), query);
+ } else {
+ pathQuery = pathQuery.chain(query);
+ }
+ }
+
+ return pathQuery;
+ }
+
+ @Override
+ public ServiceResults postItemsByQuery(ServiceContext context, Query query) throws Exception {
+ return postCollection(context);
+ }
+
+ @Override
+ public Entity updateEntity(ServiceRequest request, EntityRef ref,
+ ServicePayload payload) throws Exception {
+
+ validate(ref, payload);
+
+ Notification notification = em.get(ref, Notification.class);
+
+ if ("restart".equals(payload.getProperty("restart"))) { // for emergency
+ // use only
+ payload.getProperties().clear();
+ payload.setProperty("restart", "");
+ payload.setProperty("errorMessage", "");
+ payload.setProperty("deliver", System.currentTimeMillis() + gracePeriod);
+
+ // once finished, immutable
+ } else if (notification.getFinished() != null) {
+ throw new ForbiddenServiceOperationException(request,
+ "Notification immutable once sent.");
+
+ // once started, only cancel is allowed
+ } else if (notification.getStarted() != null) {
+ if (payload.getProperty("canceled") != Boolean.TRUE) {
+ throw new ForbiddenServiceOperationException(request,
+ "Notification has started. You may only set canceled.");
+ }
+ payload.getProperties().clear();
+ payload.setProperty("canceled", Boolean.TRUE);
+ }
+
+ Entity response = super.updateEntity(request, ref, payload);
+
+ Long deliver = (Long) payload.getProperty("deliver");
+ if (deliver != null) {
+ if (!deliver.equals(notification.getDeliver())) {
+ notificationQueueManager.processBatchAndReschedule((Notification) response, null);
+ }
+ }
+ return response;
+ }
+
+ @Override
+ protected boolean isDeleteAllowed(ServiceContext context, Entity entity) {
+ Notification notification = (Notification) entity;
+ return (notification.getStarted() == null);
+ }
+
+ public Set<String> getProviders() {
+ return providerAdapters.keySet();
+ }
+
+ // validate payloads
+ private void validate(EntityRef ref, ServicePayload servicePayload)
+ throws Exception {
+ Object obj_payloads = servicePayload.getProperty("payloads");
+ if (obj_payloads == null && ref == null) {
+ throw new RequiredPropertyNotFoundException("notification",
+ "payloads");
+ }
+ if (obj_payloads != null) {
+ if (!(obj_payloads instanceof Map)) {
+ throw new IllegalArgumentException(
+ "payloads must be a JSON Map");
+ }
+ final Map<String, Object> payloads = (Map<String, Object>) obj_payloads;
+ final Map<Object, Notifier> notifierMap = getNotifierMap(payloads);
+ Observable t = Observable.from(payloads.entrySet()).subscribeOn(Schedulers.io()).map(new Func1<Map.Entry<String, Object>, Object>() {
+ @Override
+ public Object call(Map.Entry<String, Object> entry) {
+ String notifierId = entry.getKey();
+ Notifier notifier = notifierMap.get(notifierId);
+ if (notifier == null) {
+ throw new IllegalArgumentException("notifier \""
+ + notifierId + "\" not found");
+ }
+ ProviderAdapter providerAdapter = providerAdapters.get(notifier
+ .getProvider());
+ Object payload = entry.getValue();
+ try {
+ return providerAdapter.translatePayload(payload); // validate
+ // specifically to
+ // provider
+ } catch (Exception e) {
+ return e;
+ }
+ }
+ });
+ Object e = t.toBlocking().lastOrDefault(null);
+ if(e instanceof Throwable){
+ throw new Exception((Exception)e);
+ }
+
+ }
+ }
+
+
+
+ public String getJobQueueName(EntityRef notification) {
+ return "notifications/" + notification.getUuid();
+ }
+
+
+
+ /* adds a single device for delivery - used only by tests */
+ public void addDevice(EntityRef notification, EntityRef device) throws Exception {
+
+ String jobQueueName = getJobQueueName(notification);
+ Message message = new Message();
+ message.setObjectProperty(MESSAGE_PROPERTY_DEVICE_UUID,
+ device.getUuid());
+ sm.getQueueManager().postToQueue(jobQueueName, message);
+ }
+
+ public Notification getSourceNotification(EntityRef receipt)
+ throws Exception {
+ Receipt r = em.get(receipt.getUuid(), Receipt.class);
+ return em.get(r.getNotificationUUID(), Notification.class);
+ }
+
+
+ /* create a map of Notifier UUIDs and/or names to Notifiers */
+ protected Map<Object, Notifier> getNotifierMap(Map payloads)
+ throws Exception {
+ Map<Object, Notifier> notifiers = new HashMap<Object, Notifier>(
+ payloads.size() * 2);
+ for (Object id : payloads.keySet()) {
+ Identifier identifier = Identifier.from(id);
+ Notifier notifier;
+ if (identifier.isUUID()) {
+ notifier = em.get(identifier.getUUID(), Notifier.class);
+ } else {
+ EntityRef ref = em.getAlias("notifier", identifier.getName());
+ notifier = em.get(ref, Notifier.class);
+ }
+ if (notifier != null) {
+ notifiers.put(notifier.getUuid(), notifier);
+ notifiers.put(notifier.getUuid().toString(), notifier);
+ if (notifier.getName() != null) {
+ notifiers.put(notifier.getName(), notifier);
+ }
+ }
+ }
+ return notifiers;
+ }
+
+
+ /**
+ * attempts to test the providers connections - throws an Exception on
+ * failure
+ */
+ public void testConnection(Notifier notifier) throws Exception {
+ ProviderAdapter providerAdapter = providerAdapters.get(notifier.getProvider());
+ if (providerAdapter != null) {
+ providerAdapter.testConnection(notifier);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/9d7901ae/stack/services/src/main/java/org/apache/usergrid/services/notifications/ProviderAdapter.java
----------------------------------------------------------------------
diff --git a/stack/services/src/main/java/org/apache/usergrid/services/notifications/ProviderAdapter.java b/stack/services/src/main/java/org/apache/usergrid/services/notifications/ProviderAdapter.java
new file mode 100644
index 0000000..5489c2d
--- /dev/null
+++ b/stack/services/src/main/java/org/apache/usergrid/services/notifications/ProviderAdapter.java
@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.usergrid.services.notifications;
+
+import java.util.Date;
+import java.util.Map;
+import org.apache.usergrid.persistence.EntityManager;
+import org.apache.usergrid.persistence.Notification;
+import org.apache.usergrid.persistence.Notifier;
+import org.apache.usergrid.services.ServicePayload;
+
+/**
+ * To send a Notification, the following methods should be called in this order:
+ * 1) testConnection() for each notifier to be used 2) translatePayload() for
+ * each payload to be sent 3) sendNotification() for each target 4)
+ * doneSendingNotifications() when all #2 have been called If there is an
+ * Exception in #1 or #2, you should consider the entire Notification to be
+ * invalid. Also, getInactiveDevices() should be called occasionally and may be
+ * called at any time.
+ */
+public interface ProviderAdapter {
+
+ public void testConnection(Notifier notifier) throws ConnectionException;
+
+ public void sendNotification(String providerId, Notifier notifier,
+ Object payload, Notification notification, TaskTracker tracker)
+ throws Exception;
+
+ /**
+ * must be called when done calling sendNotification() so any open batches
+ * may be closed out
+ */
+ public void doneSendingNotifications() throws Exception;
+
+ public Map<String, Date> getInactiveDevices(Notifier notifier,
+ EntityManager em) throws Exception;
+
+ public Object translatePayload(Object payload) throws Exception;
+
+ public void validateCreateNotifier(ServicePayload payload) throws Exception;
+}