Release 3.0.0

.gitignore

@@ -0,0 +1,4 @@
+target
+*.iml
+.idea
+out

AGPL_FILE_HEADER

@@ -0,0 +1,17 @@
+Copyright 2011-2016 Green Energy Corp.
+
+Licensed to Green Energy Corp (www.greenenergycorp.com) under one or more
+contributor license agreements. See the NOTICE file distributed with this
+work for additional information regarding copyright ownership. Green Energy
+Corp licenses this file to you under the GNU Affero General Public License
+Version 3.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.gnu.org/licenses/agpl.html
+
+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.
+

APACHE_FILE_HEADER

@@ -0,0 +1,16 @@
+Copyright 2011-2016 Green Energy Corp.
+
+Licensed to Green Energy Corp (www.greenenergycorp.com) under one or more
+contributor license agreements. See the NOTICE file distributed with this
+work for additional information regarding copyright ownership. Green Energy
+Corp 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.

README.md

@@ -0,0 +1,32 @@
+GreenBus Platform
+--------------------
+
+Prerequisites:
+
+- Maven 3
+- Postgresql
+- Qpid Broker (for integration tests)
+- Google Protocol Buffers compiler (version 2.5.0)
+
+### Setting up Postgresql
+
+The Postgres install must be initialized with the databases and users. The relevant configuration is contained in
+the init_postgres.sql.
+
+    sudo su postgres -c psql < init_postgres.sql
+
+### Setting up Qpid
+
+Qpid needs have auth settings that match the configuration in `io.greenbus.msg.amqp.cfg`. For development purposes,
+the Qpid auth can be disabled entirely by setting `auth=no` in the broker configuration and removing references to the
+ACL submodule.
+
+### Building with Maven
+
+Build:
+
+    mvn clean install
+
+### Dependency Projects
+
+The GreenBus platform project is based on [GreenBus Messaging](https://github.com/gec/greenbus-msg).

app-framework/pom.xml

@@ -0,0 +1,91 @@
+<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>
+
+    <parent>
+        <groupId>io.greenbus</groupId>
+        <artifactId>greenbus-scala-base</artifactId>
+        <version>3.0.0</version>
+        <relativePath>../scala-base</relativePath>
+    </parent>
+
+    <artifactId>greenbus-app-framework</artifactId>
+    <packaging>jar</packaging>
+
+    <licenses>
+        <license>
+            <name>AGPLv3</name>
+            <url>http://www.gnu.org/licenses/agpl-3.0.txt</url>
+        </license>
+    </licenses>
+
+    <build>
+        <plugins>
+
+            <plugin>
+                <groupId>com.mycila.maven-license-plugin</groupId>
+                <artifactId>maven-license-plugin</artifactId>
+                <configuration>
+                    <header>../APACHE_FILE_HEADER</header>
+                </configuration>
+            </plugin>
+            <plugin>
+                <artifactId>maven-assembly-plugin</artifactId>
+                <configuration>
+                    <descriptorRefs>
+                        <descriptorRef>jar-with-dependencies</descriptorRef>
+                    </descriptorRefs>
+                </configuration>
+                <executions>
+                    <execution>
+                        <id>make-assembly</id>
+                        <phase>package</phase>
+                        <goals>
+                            <goal>single</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
+        </plugins>
+    </build>
+
+    <dependencies>
+        <dependency>
+            <groupId>io.greenbus</groupId>
+            <artifactId>greenbus-client</artifactId>
+            <version>3.0.0</version>
+        </dependency>
+        <dependency>
+            <groupId>io.greenbus</groupId>
+            <artifactId>greenbus-util</artifactId>
+            <version>3.0.0</version>
+        </dependency>
+        <dependency>
+            <groupId>io.greenbus.msg</groupId>
+            <artifactId>greenbus-msg-qpid</artifactId>
+            <version>1.0.0</version>
+        </dependency>
+        <dependency>
+            <groupId>com.typesafe.play</groupId>
+            <artifactId>play-json_2.10</artifactId>
+            <version>2.3.9</version>
+        </dependency>
+        <dependency>
+            <groupId>com.typesafe.akka</groupId>
+            <artifactId>akka-actor_2.10</artifactId>
+            <version>2.2.0</version>
+        </dependency>
+        <dependency>
+            <groupId>com.typesafe.akka</groupId>
+            <artifactId>akka-slf4j_2.10</artifactId>
+            <version>2.2.0</version>
+        </dependency>
+        <dependency>
+            <groupId>commons-io</groupId>
+            <artifactId>commons-io</artifactId>
+            <version>2.4</version>
+        </dependency>
+    </dependencies>
+
+
+</project>

app-framework/src/main/scala/io/greenbus/app/actor/AmqpConnectionConfig.scala

@@ -0,0 +1,39 @@
+/**
+ * Copyright 2011-2016 Green Energy Corp.
+ *
+ * Licensed to Green Energy Corp (www.greenenergycorp.com) under one or more
+ * contributor license agreements. See the NOTICE file distributed with this
+ * work for additional information regarding copyright ownership. Green Energy
+ * Corp 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 io.greenbus.app.actor
+
+case class AmqpConnectionConfig(amqpConfigFileList: IndexedSeq[String],
+  failureLimit: Int,
+  retryDelayMs: Long,
+  connectionTimeoutMs: Long)
+
+object AmqpConnectionConfig {
+
+  def default(path: String): AmqpConnectionConfig = {
+    AmqpConnectionConfig(Vector(path), failureLimit = 1, retryDelayMs = 5000, connectionTimeoutMs = 10000)
+  }
+
+  def default(paths: Seq[String]): AmqpConnectionConfig = {
+    AmqpConnectionConfig(paths.toVector, failureLimit = 1, retryDelayMs = 5000, connectionTimeoutMs = 10000)
+  }
+
+  def load(paths: Seq[String], failureLimit: Int, retryDelayMs: Long, connectionTimeoutMs: Long): AmqpConnectionConfig = {
+    AmqpConnectionConfig(paths.toVector, failureLimit, retryDelayMs, connectionTimeoutMs)
+  }
+}

app-framework/src/main/scala/io/greenbus/app/actor/AmqpConnectionManager.scala

@@ -0,0 +1,179 @@
+/**
+ * Copyright 2011-2016 Green Energy Corp.
+ *
+ * Licensed to Green Energy Corp (www.greenenergycorp.com) under one or more
+ * contributor license agreements. See the NOTICE file distributed with this
+ * work for additional information regarding copyright ownership. Green Energy
+ * Corp 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 io.greenbus.app.actor
+
+import akka.actor._
+import com.typesafe.scalalogging.slf4j.Logging
+import io.greenbus.app.actor.util.NestedStateMachine
+import io.greenbus.client.ServiceConnection
+import io.greenbus.client.exception.UnauthorizedException
+import io.greenbus.msg.SessionUnusableException
+import io.greenbus.msg.amqp.AmqpSettings
+import io.greenbus.msg.qpid.QpidBroker
+import scala.concurrent.duration._
+
+object FailoverConnectionManager {
+
+  sealed trait State
+  case class Down(configOffset: Int, failures: Int, sequence: Int) extends State
+  case class Running(configOffset: Int, connection: ServiceConnection, child: ActorRef, sequence: Int) extends State
+
+  case object AttemptConnect
+  case class LostBrokerConnection(expected: Boolean, sequence: Int)
+
+  def props(processName: String,
+    amqpConfigFileList: IndexedSeq[String],
+    failureLimit: Int,
+    retryDelayMs: Long,
+    connectionTimeoutMs: Long,
+    factory: ServiceConnection => Props): Props = {
+    Props(classOf[FailoverConnectionManager], processName, amqpConfigFileList, failureLimit, retryDelayMs, connectionTimeoutMs, factory)
+  }
+}
+
+class FailoverConnectionManager(
+    processName: String,
+    amqpConfigFileList: IndexedSeq[String],
+    failureLimit: Int,
+    retryDelayMs: Long,
+    connectionTimeoutMs: Long,
+    factory: ServiceConnection => Props) extends NestedStateMachine with MessageScheduling with Logging {
+  import FailoverConnectionManager._
+
+  protected type StateType = State
+  protected def start: StateType = Down(0, 0, 0)
+
+  self ! AttemptConnect
+
+  protected def machine = {
+
+    case state @ Down(offset, failures, sequence) => {
+
+      case AttemptConnect => {
+
+        val configToAttempt = amqpConfigFileList(offset)
+        logger.info(s"Process $processName attempting connection using config: $configToAttempt")
+        logger.debug(s"Process $processName attempting connection, offset: $offset, failures: $failures, list: $amqpConfigFileList")
+
+        try {
+          val connection = connectQpid(configToAttempt)
+
+          connection.addConnectionListener { expected =>
+            self ! LostBrokerConnection(expected, sequence)
+          }
+
+          val child = context.actorOf(factory(connection))
+
+          Running(offset, connection, child, sequence)
+
+        } catch {
+          case ex: Throwable =>
+
+            logger.error(s"Couldn't initialize connection for $processName: " + ex.getMessage)
+
+            scheduleMsg(retryDelayMs, AttemptConnect)
+            if (failures >= failureLimit) {
+              val nextOffset = if (offset >= (amqpConfigFileList.size - 1)) 0 else offset + 1
+              Down(nextOffset, 0, sequence)
+            } else {
+              Down(offset, failures + 1, sequence)
+            }
+
+        }
+      }
+      case LostBrokerConnection(expected, connSeq) => {
+        if (!expected) {
+          logger.warn(s"Saw unexpected lost connection event while $processName was already down")
+        }
+        state
+      }
+    }
+
+    case state @ Running(configOffset, connection, child, sequence) => {
+
+      case LostBrokerConnection(expected, connSeq) => {
+        if (connSeq == sequence) {
+
+          try {
+            connection.disconnect()
+          } catch {
+            case ex: Throwable =>
+          }
+          child ! PoisonPill
+
+          logger.info(s"Process $processName lost connection")
+
+          scheduleMsg(retryDelayMs, AttemptConnect)
+
+          Down(configOffset, 0, sequence + 1)
+
+        } else {
+          state
+        }
+      }
+
+      case RestartConnection(reason) => {
+        logger.warn(s"Restarting $processName due to $reason")
+
+        try {
+          connection.disconnect()
+        } catch {
+          case ex: Throwable =>
+        }
+        child ! PoisonPill
+
+        scheduleMsg(retryDelayMs, AttemptConnect)
+
+        Down(configOffset, 0, sequence + 1)
+      }
+    }
+  }
+
+  private def connectQpid(path: String): ServiceConnection = {
+    val config = AmqpSettings.load(path)
+    val connection = ServiceConnection.connect(config, QpidBroker, connectionTimeoutMs)
+    logger.info(s"Connected to broker: $config")
+    connection
+  }
+
+  override protected def onShutdown(state: State): Unit = {
+    state match {
+      case Running(configOffset, connection, child, sequence) =>
+        logger.info(s"Disconnecting on shutdown for $processName")
+        connection.disconnect()
+      case _ =>
+    }
+  }
+
+  override def supervisorStrategy: SupervisorStrategy = {
+    import SupervisorStrategy._
+    OneForOneStrategy(maxNrOfRetries = 1, withinTimeRange = 3.seconds) {
+      case _: SessionUnusableException =>
+        self ! RestartConnection("session error")
+        Stop
+      case _: UnauthorizedException =>
+        self ! RestartConnection("auth failure")
+        Stop
+      case ex: Throwable =>
+        self ! RestartConnection("unknown error: " + ex)
+        Stop
+    }
+  }
+
+}