Permalink
Browse files

Refactoring JmsStages to allow confirgurable recovery time for sessio…

…ns that have been closed

after an exception.
  • Loading branch information...
atooni committed Jan 16, 2019
1 parent a415cd6 commit 6fe506f4e7b866feb8b7adfcf2191e3698dce343
Showing with 257 additions and 192 deletions.
  1. +1 −0 blended.activemq.client/src/main/scala/blended/activemq/client/RoundtripConnectionVerifier.scala
  2. +2 −0 ....akka.http.jmsqueue/src/test/scala/blended/akka/http/jmsqueue/internal/HttpQueueServiceSpec.scala
  3. +2 −0 ....akka.http.jmsqueue/src/test/scala/blended/akka/http/jmsqueue/internal/SttpQueueServiceSpec.scala
  4. +9 −6 blended.jms.bridge/src/main/scala/blended/jms/bridge/JmsStreamBuilder.scala
  5. +5 −2 blended.jms.bridge/src/main/scala/blended/jms/bridge/internal/BridgeController.scala
  6. +7 −2 blended.jms.bridge/src/main/scala/blended/jms/bridge/internal/InboundConfig.scala
  7. +4 −1 blended.jms.bridge/src/test/scala/blended/jms/bridge/internal/BridgeSpec.scala
  8. +25 −27 blended.launcher/src/main/scala/blended/launcher/jvmrunner/JvmLauncher.scala
  9. +3 −3 blended.launcher/src/main/scala/blended/launcher/jvmrunner/RunningProcess.scala
  10. +15 −7 blended.launcher/src/runner/resources/bin/blended.sh
  11. +13 −6 blended.launcher/src/runner/resources/bin/install-service.bat
  12. +13 −0 blended.launcher/src/test/scala/blended/launcher/jvmrunner/RunningProcessSpec.scala
  13. +9 −0 blended.security.ssl/src/test/scala/blended/security/ssl/internal/QuickSSLTests.scala
  14. +2 −2 ...d.streams.dispatcher/src/main/scala/blended/streams/dispatcher/internal/builder/CbeSendFlow.scala
  15. +6 −6 ...ms.dispatcher/src/main/scala/blended/streams/dispatcher/internal/builder/RunnableDispatcher.scala
  16. +2 −2 ...s.dispatcher/src/main/scala/blended/streams/dispatcher/internal/builder/TransactionOutbound.scala
  17. +2 −0 ...reams.dispatcher/src/test/scala/blended/streams/dispatcher/internal/DispatcherActivatorSpec.scala
  18. +1 −0 ...spatcher/src/test/scala/blended/streams/dispatcher/internal/builder/TransactionOutboundSpec.scala
  19. +1 −0 blended.streams.testsupport/src/main/scala/blended/streams/testsupport/RoundtripHelper.scala
  20. +41 −7 blended.streams/src/main/scala/blended/streams/jms/JMSConnector.scala
  21. +28 −50 blended.streams/src/main/scala/blended/streams/jms/JmsAckSourceStage.scala
  22. +20 −6 blended.streams/src/main/scala/blended/streams/jms/JmsSettings.scala
  23. +16 −14 blended.streams/src/main/scala/blended/streams/jms/JmsSinkStage.scala
  24. +1 −17 blended.streams/src/main/scala/blended/streams/jms/JmsSourceStage.scala
  25. +8 −9 blended.streams/src/main/scala/blended/streams/jms/JmsStageLogic.scala
  26. +9 −14 blended.streams/src/main/scala/blended/streams/jms/JmsStreamSupport.scala
  27. +2 −4 blended.streams/src/main/scala/blended/streams/jms/SourceStageLogic.scala
  28. +2 −2 blended.streams/src/main/scala/blended/streams/transaction/TransactionWiretap.scala
  29. +3 −3 blended.streams/src/test/scala/blended/streams/jms/JmsAckSourceSpec.scala
  30. +5 −2 project/BlendedLauncher.scala
@@ -53,6 +53,7 @@ class RoundtripConnectionVerifier(
.withHeader(replyToHeader(headerConfig.prefix), responseDest.asString).get

val pSettings : JmsProducerSettings = JmsProducerSettings(
log = log,
connectionFactory = cf,
jmsDestination = Some(requestDest),
destinationResolver = s => new MessageDestinationResolver(headerConfig, s)
@@ -82,6 +82,7 @@ class HttpQueueServiceSpec extends FreeSpec
val env : FlowEnvelope = FlowEnvelope(FlowMessage(msg)(FlowMessage.noProps))

val pSettings : JmsProducerSettings = JmsProducerSettings(
log = log,
connectionFactory = cf,
jmsDestination = Some(JmsQueue("Queue1"))
)
@@ -108,6 +109,7 @@ class HttpQueueServiceSpec extends FreeSpec
val env : FlowEnvelope = FlowEnvelope(FlowMessage(ByteString(msg))(FlowMessage.noProps))

val pSettings : JmsProducerSettings = JmsProducerSettings(
log = log,
connectionFactory = cf,
jmsDestination = Some(JmsQueue("Queue1"))
)
@@ -90,6 +90,7 @@ class SttpQueueServiceSpec extends SimplePojoContainerSpec
val env : FlowEnvelope = FlowEnvelope(FlowMessage(msg)(FlowMessage.noProps))

val pSettings : JmsProducerSettings = JmsProducerSettings(
log = log,
connectionFactory = amqCF,
jmsDestination = Some(JmsQueue("Queue1"))
)
@@ -111,6 +112,7 @@ class SttpQueueServiceSpec extends SimplePojoContainerSpec
val env : FlowEnvelope = FlowEnvelope(FlowMessage(ByteString(msg))(FlowMessage.noProps))

val pSettings : JmsProducerSettings = JmsProducerSettings(
log = log,
connectionFactory = amqCF,
jmsDestination = Some(JmsQueue("Queue1"))
)
@@ -15,6 +15,7 @@ import blended.streams.{FlowProcessor, StreamControllerConfig}
import blended.util.logging.Logger
import com.typesafe.config.Config

import scala.concurrent.duration.FiniteDuration
import scala.util.Try

class InvalidBridgeConfigurationException(msg: String) extends Exception(msg)
@@ -38,7 +39,8 @@ case class JmsStreamConfig(
subscriberName : Option[String],
header : List[HeaderProcessorConfig],
idSvc : Option[ContainerIdentifierService] = None,
rawConfig : Config
rawConfig : Config,
sessionRecreateTimeout : FiniteDuration
)

class JmsStreamBuilder(
@@ -49,9 +51,10 @@ class JmsStreamBuilder(
private val inId = s"${cfg.fromCf.vendor}:${cfg.fromCf.provider}:${cfg.fromDest.asString}"
private val outId = s"${cfg.toCf.vendor}:${cfg.toCf.provider}:${cfg.toDest.map(_.asString).getOrElse("out")}"
private val streamId = s"${cfg.headerCfg.prefix}.bridge.JmsStream($inId->$outId)"
private val bridgeLogger = Logger(streamId)

// configure the consumer
private val srcSettings = JMSConsumerSettings(cfg.fromCf)
private val srcSettings = JMSConsumerSettings(bridgeLogger, cfg.fromCf)
.withAcknowledgeMode(AcknowledgeMode.ClientAcknowledge)
.withDestination(Some(cfg.fromDest))
.withSessionCount(cfg.listener)
@@ -68,13 +71,14 @@ class JmsStreamBuilder(
}

private val toSettings = JmsProducerSettings(
log = bridgeLogger,
connectionFactory = cfg.toCf
)
.withDestination(cfg.toDest)
.withDestinationResolver(destResolver)
.withDeliveryMode(JmsDeliveryMode.Persistent)

private val bridgeLogger = Logger(streamId)


private val internalProvider : Try[BridgeProviderConfig] = cfg.registry.internalProvider
private val internalId = (internalProvider.get.vendor, internalProvider.get.provider)
@@ -136,8 +140,7 @@ class JmsStreamBuilder(
Source.fromGraph(new JmsAckSourceStage(
name = streamId + "-source",
settings = srcSettings,
headerConfig = cfg.headerCfg,
log = bridgeLogger
headerConfig = cfg.headerCfg
))

val jmsSource : Source[FlowEnvelope, NotUsed] = if (cfg.inbound && cfg.header.nonEmpty) {
@@ -159,7 +162,7 @@ class JmsStreamBuilder(

jmsSource
.via(Flow.fromGraph(g))
.via(jmsProducer(name = streamId + "-sink", settings = toSettings, autoAck = true, log = bridgeLogger))
.via(jmsProducer(name = streamId + "-sink", settings = toSettings, autoAck = true))
}

bridgeLogger.info(s"Starting bridge stream with config [inbound=${cfg.inbound},trackTransaction=${cfg.trackTransaction}]")
@@ -14,6 +14,7 @@ import com.typesafe.config.Config

import scala.collection.JavaConverters._
import scala.util.{Failure, Success}
import scala.concurrent.duration._

private[bridge] object BridgeControllerConfig {

@@ -101,7 +102,8 @@ class BridgeController(ctrlCfg: BridgeControllerConfig)(implicit system : ActorS
subscriberName = in.subscriberName,
header = in.header,
idSvc = Some(ctrlCfg.idSvc),
rawConfig = ctrlCfg.rawConfig
rawConfig = ctrlCfg.rawConfig,
sessionRecreateTimeout = in.sessionRecreateTimeout
)

val streamCfg: StreamControllerConfig = new JmsStreamBuilder(inCfg).streamCfg
@@ -133,7 +135,8 @@ class BridgeController(ctrlCfg: BridgeControllerConfig)(implicit system : ActorS
trackTransaction = TrackTransaction.FromMessage,
subscriberName = None,
header = List.empty,
rawConfig = ctrlCfg.rawConfig
rawConfig = ctrlCfg.rawConfig,
sessionRecreateTimeout = 1.second
)

val streamCfg: StreamControllerConfig = new JmsStreamBuilder(outCfg).streamCfg
@@ -7,6 +7,7 @@ import blended.streams.processor.HeaderProcessorConfig
import blended.util.config.Implicits._
import com.typesafe.config.Config

import scala.concurrent.duration._
import scala.util.Try

object InboundConfig {
@@ -39,6 +40,8 @@ object InboundConfig {
HeaderProcessorConfig.create(cfg)
}

val sessionRecreateTimeout : FiniteDuration = cfg.getDuration("sessionRecreateTimeout", 1.second)

InboundConfig(
name = name,
vendor = vendor,
@@ -48,7 +51,8 @@ object InboundConfig {
persistent = persistent,
subscriberName = subscriberName,
listener = listener,
header = header
header = header,
sessionRecreateTimeout = sessionRecreateTimeout
)
}
}
@@ -62,5 +66,6 @@ case class InboundConfig (
persistent : JmsDeliveryMode,
subscriberName : Option[String],
listener : Int,
header : List[HeaderProcessorConfig]
header : List[HeaderProcessorConfig],
sessionRecreateTimeout : FiniteDuration
)
@@ -74,7 +74,8 @@ class BridgeSpec extends SimplePojoContainerSpec
trackTransaction = TrackTransaction.Off,
subscriberName = None,
header = List.empty,
rawConfig = ctrlCfg.rawConfig
rawConfig = ctrlCfg.rawConfig,
sessionRecreateTimeout = 1.second
)

private val streamCfg = new JmsStreamBuilder(cfg).streamCfg
@@ -101,6 +102,7 @@ class BridgeSpec extends SimplePojoContainerSpec
} map { FlowEnvelope.apply }

val pSettings : JmsProducerSettings = JmsProducerSettings(
log = log,
connectionFactory = external,
jmsDestination = Some(JmsQueue("sampleIn"))
)
@@ -148,6 +150,7 @@ class BridgeSpec extends SimplePojoContainerSpec
).get))

val pSettings : JmsProducerSettings = JmsProducerSettings(
log = log,
connectionFactory = external,
jmsDestination = Some(JmsQueue("SampleHeaderIn"))
)
@@ -1,16 +1,17 @@
package blended.launcher.jvmrunner

import java.io.{ File, FileInputStream }
import java.io.{File, FileInputStream}
import java.util.Properties

import scala.collection.JavaConverters._
import scala.util.Try
import scala.util.control.NonFatal

import blended.launcher.internal.ARM
import blended.updater.config.OverlayConfigCompanion
import blended.util.logging.Logger

import scala.annotation.tailrec

object JvmLauncher {

private[this] lazy val log = Logger[JvmLauncher.type]
@@ -28,7 +29,7 @@ object JvmLauncher {
}

/**
* A small Java wrapper rresponsiblefor controlling the actual Container JVM.
* A small Java wrapper responsiblefor controlling the actual Container JVM.
*/
class JvmLauncher() {

@@ -166,7 +167,7 @@ class JvmLauncher() {
}
}

case class Config(
case class JvmLauncherConfig(
classpath: Seq[File] = Seq(),
otherArgs: Seq[String] = Seq(),
action: Option[String] = None,
@@ -186,7 +187,9 @@ class JvmLauncher() {
override def toString(): String = prettyPrint
}

def parse(args: Seq[String], initialConfig: Config = Config()): Config = {
@tailrec
final def parse(args: Seq[String], initialConfig: JvmLauncherConfig = JvmLauncherConfig()): JvmLauncherConfig = {

args match {
case Seq() =>
initialConfig
@@ -214,13 +217,18 @@ class JvmLauncher() {
}
}

def checkConfig(config: Config): Try[Config] = Try {
if (config.action.isEmpty) sys.error("Missing arguments for action: start|stop")
if (config.classpath.isEmpty) Console.err.println("Warning: No classpath given")
private def checkConfig(config: JvmLauncherConfig): Try[JvmLauncherConfig] = Try {
if (config.action.isEmpty) {
sys.error("Missing arguments for action: start|stop")
}

if (config.classpath.isEmpty) {
Console.err.println("Warning: No classpath given")
}
config
}

def startJava(classpath: Seq[File],
private def startJava(classpath: Seq[File],
jvmOpts: Array[String],
arguments: Array[String],
interactive: Boolean = false,
@@ -231,30 +239,20 @@ class JvmLauncher() {
log.debug("About to run Java process")

// lookup java by JAVA_HOME env variable
val javaHome = System.getenv("JAVA_HOME")
val java =
if (javaHome != null) s"${
javaHome
}/bin/java"
else "java"
val java = Option(System.getenv("JAVA_HOME")) match {
case Some(javaHome) => s"$javaHome/bin/java"
case None => "java"
}

log.debug("Using java executable: " + java)

val cpArgs = classpath match {
case null | Seq() => Array[String]()
case cp => Array("-cp", pathAsArg(classpath))
val cpArgs = Option(classpath) match {
case None | Some(Seq()) => Array[String]()
case Some(cp) => Array("-cp", pathAsArg(classpath))
}
log.debug("Using classpath args: " + cpArgs.mkString(" "))

// FIXME: Only pass explicitly configured System properties to the inner JVM
val propArgs = System.getProperties.asScala.map(p => s"-D${
p._1
}=${
p._2
}").toArray[String]
log.debug("Using property args: " + propArgs.mkString(" "))

val command = Array(java) ++ cpArgs ++ jvmOpts ++ propArgs ++ arguments
val command = Array(java) ++ cpArgs ++ jvmOpts ++ arguments

val pb = new ProcessBuilder(command: _*)
log.debug("Run command: " + command.mkString(" "))
@@ -64,7 +64,7 @@ class RunningProcess(process: Process, errorsIntoOutput: Boolean, interactive: B
* Starts a new thread which copies an InputStream into an Output stream. Does not close the streams.
*/

def asyncCopy(in: InputStream, out: OutputStream, immediately: Boolean = false): Thread =
private def asyncCopy(in: InputStream, out: OutputStream, immediately: Boolean = false): Thread =
new Thread("StreamCopyThread") {
setDaemon(true)

@@ -84,7 +84,7 @@ class RunningProcess(process: Process, errorsIntoOutput: Boolean, interactive: B
/**
* Copies an InputStream into an OutputStream. Does not close the streams.
*/
def copy(in: InputStream, out: OutputStream, immediately: Boolean = false): Unit = {
private def copy(in: InputStream, out: OutputStream, immediately: Boolean = false): Unit = {
if (immediately) {
while (true) {
if (in.available > 0) {
@@ -109,4 +109,4 @@ class RunningProcess(process: Process, errorsIntoOutput: Boolean, interactive: B
}
}
}
}
}
@@ -44,15 +44,23 @@ if [ "x$BLENDED_STRICT" != "x" ] ; then
LAUNCHER_OPTS="$LAUNCHER_OPTS --strict"
fi

LOGBACK_CONFIG_SETTING="-Dlogback.configurationFile=${BLENDED_HOME}/etc/logback.xml"

# Options for the service daemen JVM (outer) with controls the container JVM
JAVA_OPTS="${JAVA_OPTS} -Xmx24m"
JAVA_OPTS="${JAVA_OPTS} -Dlogback.configurationFile=${BLENDED_HOME}/etc/logback.xml"
JAVA_OPTS="${JAVA_OPTS} ${LOGBACK_CONFIG_SETTING}"
JAVA_OPTS="${JAVA_OPTS} -Dblended.home=${BLENDED_HOME}"
JAVA_OPTS="${JAVA_OPTS} -Dsun.net.client.defaultConnectTimeout=500 -Dsun.net.client.defaultReadTimeout=500"

# Options for the container JVM (inner) started/managed by the service daemon JVM
# Use prefix "-jvmOpt=" to mark JVM options for the container JVM
#CONTAINER_JAVA_OPTS="${CONTAINER_JAVA_OPTS} -jvmOpt=-Xmx1024m"
# Use prefix "-jvmOpt=" to mark each JVM option to be passed to the container JVM

CONTAINER_JAVA_OPTS="${CONTAINER_JAVA_OPTS} -jvmOpt=-Dsun.net.client.defaultConnectTimeout=500"
CONTAINER_JAVA_OPTS="${CONTAINER_JAVA_OPTS} -jvmOpt=-Dsun.net.client.defaultReadTimeout=500"
CONTAINER_JAVA_OPTS="${CONTAINER_JAVA_OPTS} -jvmOpt=${LOGBACK_CONFIG_SETTING}"
CONTAINER_JAVA_OPTS="${CONTAINER_JAVA_OPTS} -jvmOpt=-Dblended.home=${BLENDED_HOME}"

# Enable this when you need to debug SSL issues
CONTAINER_JAVA_OPTS="${CONTAINER_JAVA_OPTS} -jvmOpt=-Djavax.net.debug=ssl"

if [ -n "$DEBUG_PORT" ] ; then
if [ -n "$DEBUG_WAIT" ] ; then
@@ -79,11 +87,11 @@ OUTER_CP="${BLENDED_HOME}/lib/*"
# semicolon-separated
INNER_CP="\
${BLENDED_HOME}/etc;\
${BLENDED_HOME}/lib/blended.launcher-@blended.launcher.version@.jar;\
${BLENDED_HOME}/lib/blended.launcher_@scala.binary.version@-@blended.launcher.version@.jar;\
${BLENDED_HOME}/lib/config-@typesafe.config.version@.jar;\
${BLENDED_HOME}/lib/org.osgi.core-@org.osgi.core.version@.jar;\
${BLENDED_HOME}/lib/blended.updater.config-@blended.updater.config.version@.jar;\
${BLENDED_HOME}/lib/blended.util.logging-@blended.util.logging.version@.jar;\
${BLENDED_HOME}/lib/blended.updater.config_@scala.binary.version@-@blended.updater.config.version@.jar;\
${BLENDED_HOME}/lib/blended.util.logging_@scala.binary.version@-@blended.util.logging.version@.jar;\
${BLENDED_HOME}/lib/de.tototec.cmdoption-@cmdoption.version@.jar;\
${BLENDED_HOME}/lib/scala-library-@scala.library.version@.jar;\
${BLENDED_HOME}/lib/slf4j-api-@slf4j.version@.jar;\
Oops, something went wrong.

0 comments on commit 6fe506f

Please sign in to comment.