Skip to content

Commit 58c8f64

Browse files
comcast-jonmcomcast-jonm
committed
Move SiriusFactory to top-level api package.
This is a re-spin of Comcast#27 but also incorporates the intent of Comcast#68. This should basically mean that a client of the Sirius library ought only to directly use classes, objects, and traits from the `com.comcast.xfinity.sirius.api` package, and that, going forward, those are the files that must retain backwards compatibility. The main changes involved are: * Additional extension methods for 1.3 are now collected in the Sirius1Dot3 trait that extends the existing Sirius trait. * Existing SiriusFactory was hoisted up to api package and now returns a Sirius1Dot3 instead of a SiriusImpl. Clients ought to be depending on a trait rather than a concrete class here. * There is still a SiriusFactory in the old location, but it just delegates to the one in the new location, then downcasts the return result to a SiriusImpl. This SiriusFactory is marked as deprecated.
1 parent 225dd32 commit 58c8f64

File tree

5 files changed

+229
-146
lines changed

5 files changed

+229
-146
lines changed

‎src/main/scala/com/comcast/xfinity/sirius/api/Sirius.scala

+5
Original file line numberDiff line numberDiff line change
@@ -41,3 +41,8 @@ trait Sirius {
4141
*/
4242
def isOnline: Boolean
4343
}
44+
45+
trait Sirius1Dot3 extends Sirius {
46+
def shutdown(): Unit
47+
}
48+

‎src/main/scala/com/comcast/xfinity/sirius/api/SiriusFactory.scala

+209
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,209 @@
1+
/*
2+
* Copyright 2012-2014 Comcast Cable Communications Management, LLC
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
package com.comcast.xfinity.sirius.api
17+
18+
import java.io.File
19+
import java.lang.management.ManagementFactory
20+
import java.net.InetAddress
21+
import java.util.{HashMap => JHashMap}
22+
import javax.management.ObjectName
23+
24+
import com.comcast.xfinity.sirius.admin.ObjectNameHelper
25+
import com.comcast.xfinity.sirius.api.impl.SiriusImpl
26+
import com.comcast.xfinity.sirius.info.SiriusInfo
27+
import com.comcast.xfinity.sirius.uberstore.UberStore
28+
import com.comcast.xfinity.sirius.uberstore.segmented.SegmentedUberStore
29+
import com.comcast.xfinity.sirius.util.AkkaExternalAddressResolver
30+
import com.comcast.xfinity.sirius.writeaheadlog.{CachedSiriusLog, SiriusLog}
31+
import com.typesafe.config.{Config, ConfigFactory}
32+
import akka.actor.{ActorRef, ActorSystem}
33+
import org.slf4j.LoggerFactory
34+
35+
import scala.collection.JavaConverters._
36+
37+
/**
38+
* Provides the factory for [[com.comcast.xfinity.sirius.api.impl.SiriusImpl]] instances
39+
*/
40+
object SiriusFactory {
41+
val traceLog = LoggerFactory.getLogger("SiriusFactory")
42+
43+
/**
44+
* SiriusImpl factory method, takes parameters to construct a SiriusImplementation and the dependent
45+
* ActorSystem and return the created instance. Calling shutdown on the produced SiriusImpl will also
46+
* shutdown the dependent ActorSystem.
47+
*
48+
* @param requestHandler the RequestHandler containing callbacks for manipulating the system's state
49+
* @param siriusConfig a SiriusConfiguration containing configuration info needed for this node.
50+
* @see SiriusConfiguration for info on needed config.
51+
*
52+
* @return A SiriusImpl constructed using the parameters
53+
*/
54+
def createInstance(requestHandler: RequestHandler, siriusConfig: SiriusConfiguration): Sirius1Dot3 = {
55+
val uberStoreDir = siriusConfig.getProp[String](SiriusConfiguration.LOG_LOCATION) match {
56+
case Some(dir) => dir
57+
case None =>
58+
throw new IllegalArgumentException(SiriusConfiguration.LOG_LOCATION + " must be set on config")
59+
}
60+
61+
val backendLog = {
62+
siriusConfig.getProp(SiriusConfiguration.LOG_VERSION_ID, "") match {
63+
case version if version == SegmentedUberStore.versionId => SegmentedUberStore(uberStoreDir, siriusConfig)
64+
case _ => UberStore(uberStoreDir)
65+
}
66+
}
67+
68+
val log: SiriusLog = {
69+
if (siriusConfig.getProp(SiriusConfiguration.LOG_USE_WRITE_CACHE, true)) {
70+
val cacheSize = siriusConfig.getProp(SiriusConfiguration.LOG_WRITE_CACHE_SIZE, 10000)
71+
CachedSiriusLog(backendLog, cacheSize)
72+
} else {
73+
backendLog
74+
}
75+
}
76+
77+
createInstance(requestHandler, siriusConfig, log)
78+
}
79+
80+
/**
81+
* USE ONLY FOR TESTING HOOK WHEN YOU NEED TO MOCK OUT A LOG.
82+
* Real code should use the two argument factory method.
83+
*
84+
* @param requestHandler the RequestHandler containing callbacks for manipulating the system's state
85+
* @param siriusConfig a SiriusConfiguration containing configuration info needed for this node.
86+
* @see SiriusConfiguration for info on needed config.
87+
* @param siriusLog the persistence layer to which events should be committed to and replayed from.
88+
*
89+
* @return A SiriusImpl constructed using the parameters
90+
*/
91+
private[sirius] def createInstance(requestHandler: RequestHandler, siriusConfig: SiriusConfiguration,
92+
siriusLog: SiriusLog): Sirius1Dot3 = {
93+
94+
val systemName = siriusConfig.getProp(SiriusConfiguration.AKKA_SYSTEM_NAME, "sirius-system")
95+
96+
implicit val actorSystem = ActorSystem(systemName, createActorSystemConfig(siriusConfig))
97+
98+
// inject an mbean server, without regard for the one that may have been there
99+
val mbeanServer = ManagementFactory.getPlatformMBeanServer
100+
siriusConfig.setProp(SiriusConfiguration.MBEAN_SERVER, mbeanServer)
101+
102+
// inject AkkaExternalAddressResolver
103+
siriusConfig.setProp(SiriusConfiguration.AKKA_EXTERNAL_ADDRESS_RESOLVER, AkkaExternalAddressResolver(actorSystem) (siriusConfig))
104+
105+
// here it is! the real deal creation
106+
val impl = SiriusImpl(requestHandler, siriusLog, siriusConfig)
107+
108+
// create a SiriusInfo MBean which will remain registered until we explicitly shutdown sirius
109+
val (siriusInfoObjectName, siriusInfo) = createSiriusInfoMBean(actorSystem, impl.supervisor)(siriusConfig)
110+
mbeanServer.registerMBean(siriusInfo, siriusInfoObjectName)
111+
112+
// need to shut down the actor system and unregister the mbeans when sirius is done
113+
impl.onShutdown({
114+
actorSystem.shutdown()
115+
actorSystem.awaitTermination()
116+
mbeanServer.unregisterMBean(siriusInfoObjectName)
117+
})
118+
119+
impl
120+
}
121+
122+
private def createSiriusInfoMBean(actorSystem: ActorSystem, siriusSup: ActorRef)
123+
(siriusConfig: SiriusConfiguration): (ObjectName, SiriusInfo) = {
124+
val resolver = siriusConfig.getProp[AkkaExternalAddressResolver](SiriusConfiguration.AKKA_EXTERNAL_ADDRESS_RESOLVER).
125+
getOrElse(throw new IllegalStateException("SiriusConfiguration.AKKA_EXTERNAL_ADDRESS_RESOLVER returned nothing"))
126+
val siriusInfo = new SiriusInfo(actorSystem, siriusSup, resolver)
127+
val objectNameHelper = new ObjectNameHelper
128+
val siriusInfoObjectName = objectNameHelper.getObjectName(siriusInfo, siriusSup, actorSystem)(siriusConfig)
129+
(siriusInfoObjectName, siriusInfo)
130+
}
131+
132+
/**
133+
* Creates configuration for the ActorSystem. The config precedence is as follows:
134+
* 1) host/port config trump all
135+
* 2) siriusConfig supplied external config next
136+
* 3) sirius-akka-base.conf, packaged with sirius, loaded with ConfigFactory.load
137+
*/
138+
private def createActorSystemConfig(siriusConfig: SiriusConfiguration): Config = {
139+
val hostPortConfig = createHostPortConfig(siriusConfig)
140+
val externalConfig = createExternalConfig(siriusConfig)
141+
val baseAkkaConfig = ConfigFactory.load("sirius-akka-base.conf")
142+
143+
hostPortConfig.withFallback(externalConfig).withFallback(baseAkkaConfig)
144+
}
145+
146+
private def createHostPortConfig(siriusConfig: SiriusConfiguration): Config = {
147+
val configMap = new JHashMap[String, Any]()
148+
val sslEnabled = siriusConfig.getProp(SiriusConfiguration.ENABLE_SSL, false)
149+
val transportPrefix = if (sslEnabled) "akka.remote.netty.ssl" else "akka.remote.netty.tcp"
150+
traceLog.info(s"AKKA using transport: $transportPrefix")
151+
152+
configMap.put("akka.remote.enabled-transports", List(transportPrefix).asJava)
153+
configMap.put(s"$transportPrefix.hostname",
154+
siriusConfig.getProp(SiriusConfiguration.HOST, InetAddress.getLocalHost.getHostName))
155+
configMap.put(s"$transportPrefix.port", siriusConfig.getProp(SiriusConfiguration.PORT, 2552))
156+
157+
val maxMessageSize = siriusConfig.getProp(SiriusConfiguration.MAX_AKKA_MESSAGE_SIZE_KB, "1024")
158+
val bufferSize = maxMessageSize * 2
159+
configMap.put(s"$transportPrefix.maximum-frame-size", s"${maxMessageSize}k")
160+
configMap.put(s"$transportPrefix.send-buffer-size", s"${bufferSize}k")
161+
configMap.put(s"$transportPrefix.receive-buffer-size", s"${bufferSize}k")
162+
163+
if (sslEnabled) {
164+
configMap.put(s"$transportPrefix.random-number-generator",
165+
siriusConfig.getProp(SiriusConfiguration.SSL_RANDOM_NUMBER_GENERATOR, ""))
166+
167+
configMap.put(s"$transportPrefix.security.key-store",
168+
siriusConfig.getProp(SiriusConfiguration.KEY_STORE_LOCATION,
169+
throw new IllegalArgumentException("No key-store provided")))
170+
171+
configMap.put(s"$transportPrefix.security.trust-store",
172+
siriusConfig.getProp(SiriusConfiguration.TRUST_STORE_LOCATION,
173+
throw new IllegalArgumentException("No trust-store provided")))
174+
175+
configMap.put(s"$transportPrefix.security.key-store-password",
176+
siriusConfig.getProp(SiriusConfiguration.KEY_STORE_PASSWORD,
177+
throw new IllegalArgumentException("No key-store-password value provided")))
178+
179+
configMap.put(s"$transportPrefix.security.key-password",
180+
siriusConfig.getProp(SiriusConfiguration.KEY_PASSWORD,
181+
throw new IllegalArgumentException("No key-password value provided")))
182+
183+
configMap.put(s"$transportPrefix.security.trust-store-password",
184+
siriusConfig.getProp(SiriusConfiguration.TRUST_STORE_PASSWORD,
185+
throw new IllegalArgumentException("No trust-store-password value provided")))
186+
}
187+
188+
// this is just so that the intellij shuts up
189+
ConfigFactory.parseMap(configMap.asInstanceOf[JHashMap[String, _ <: AnyRef]])
190+
}
191+
192+
/**
193+
* If siriusConfig is such configured, will load up an external configuration
194+
* for the Akka ActorSystem which is created. The filesystem is checked first,
195+
* then the classpath, if neither exist, or siriusConfig is not configured as
196+
* much, then an empty Config object is returned.
197+
*/
198+
private def createExternalConfig(siriusConfig: SiriusConfiguration): Config =
199+
siriusConfig.getProp[String](SiriusConfiguration.AKKA_EXTERN_CONFIG) match {
200+
case None => ConfigFactory.empty()
201+
case Some(externConfig) =>
202+
val externConfigFile = new File(externConfig)
203+
if (externConfigFile.exists()) {
204+
ConfigFactory.parseFile(externConfigFile).resolve()
205+
} else {
206+
ConfigFactory.parseResources(externConfig).resolve()
207+
}
208+
}
209+
}

‎src/main/scala/com/comcast/xfinity/sirius/api/impl/SiriusFactory.scala

+10-140
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@ import java.util.{HashMap => JHashMap}
2323
import com.comcast.xfinity.sirius.admin.ObjectNameHelper
2424
import com.comcast.xfinity.sirius.api.RequestHandler
2525
import com.comcast.xfinity.sirius.api.SiriusConfiguration
26+
import com.comcast.xfinity.sirius.api.{SiriusFactory => NewSiriusFactory}
2627
import com.comcast.xfinity.sirius.info.SiriusInfo
2728
import com.comcast.xfinity.sirius.writeaheadlog.CachedSiriusLog
2829
import com.comcast.xfinity.sirius.writeaheadlog.SiriusLog
@@ -39,11 +40,15 @@ import scala.collection.JavaConverters._
3940
import org.slf4j.LoggerFactory
4041

4142
/**
42-
* Provides the factory for [[com.comcast.xfinity.sirius.api.impl.SiriusImpl]] instances
43+
* Provides the factory for [[com.comcast.xfinity.sirius.api.impl.SiriusImpl]] instances. Please note that
44+
* this is no longer the recommended way of creating a Sirius implementation; this factory ought to have
45+
* lived in the top-level API package, and really ought to have returned a trait rather than a concrete.
46+
* This implementation is preserved for backwards compatibility, although it delegates to a corrected
47+
* [[com.comcast.xfinity.sirius.api.SiriusFactory]].
4348
*/
49+
@deprecated(message = "Please use com.comcast.xfinity.sirius.api.SiriusFactory instead",
50+
since = "1.3.0")
4451
object SiriusFactory {
45-
val traceLog = LoggerFactory.getLogger("SiriusFactory")
46-
4752
/**
4853
* SiriusImpl factory method, takes parameters to construct a SiriusImplementation and the dependent
4954
* ActorSystem and return the created instance. Calling shutdown on the produced SiriusImpl will also
@@ -56,29 +61,7 @@ object SiriusFactory {
5661
* @return A SiriusImpl constructed using the parameters
5762
*/
5863
def createInstance(requestHandler: RequestHandler, siriusConfig: SiriusConfiguration): SiriusImpl = {
59-
val uberStoreDir = siriusConfig.getProp[String](SiriusConfiguration.LOG_LOCATION) match {
60-
case Some(dir) => dir
61-
case None =>
62-
throw new IllegalArgumentException(SiriusConfiguration.LOG_LOCATION + " must be set on config")
63-
}
64-
65-
val backendLog = {
66-
siriusConfig.getProp(SiriusConfiguration.LOG_VERSION_ID, "") match {
67-
case version if version == SegmentedUberStore.versionId => SegmentedUberStore(uberStoreDir, siriusConfig)
68-
case _ => UberStore(uberStoreDir)
69-
}
70-
}
71-
72-
val log: SiriusLog = {
73-
if (siriusConfig.getProp(SiriusConfiguration.LOG_USE_WRITE_CACHE, true)) {
74-
val cacheSize = siriusConfig.getProp(SiriusConfiguration.LOG_WRITE_CACHE_SIZE, 10000)
75-
CachedSiriusLog(backendLog, cacheSize)
76-
} else {
77-
backendLog
78-
}
79-
}
80-
81-
createInstance(requestHandler, siriusConfig, log)
64+
NewSiriusFactory.createInstance(requestHandler, siriusConfig).asInstanceOf[SiriusImpl]
8265
}
8366

8467
/**
@@ -94,120 +77,7 @@ object SiriusFactory {
9477
*/
9578
private[sirius] def createInstance(requestHandler: RequestHandler, siriusConfig: SiriusConfiguration,
9679
siriusLog: SiriusLog): SiriusImpl = {
97-
98-
val systemName = siriusConfig.getProp(SiriusConfiguration.AKKA_SYSTEM_NAME, "sirius-system")
99-
100-
implicit val actorSystem = ActorSystem(systemName, createActorSystemConfig(siriusConfig))
101-
102-
// inject an mbean server, without regard for the one that may have been there
103-
val mbeanServer = ManagementFactory.getPlatformMBeanServer
104-
siriusConfig.setProp(SiriusConfiguration.MBEAN_SERVER, mbeanServer)
105-
106-
// inject AkkaExternalAddressResolver
107-
siriusConfig.setProp(SiriusConfiguration.AKKA_EXTERNAL_ADDRESS_RESOLVER, AkkaExternalAddressResolver(actorSystem) (siriusConfig))
108-
109-
// here it is! the real deal creation
110-
val impl = SiriusImpl(requestHandler, siriusLog, siriusConfig)
111-
112-
// create a SiriusInfo MBean which will remain registered until we explicitly shutdown sirius
113-
val (siriusInfoObjectName, siriusInfo) = createSiriusInfoMBean(actorSystem, impl.supervisor)(siriusConfig)
114-
mbeanServer.registerMBean(siriusInfo, siriusInfoObjectName)
115-
116-
// need to shut down the actor system and unregister the mbeans when sirius is done
117-
impl.onShutdown({
118-
actorSystem.shutdown()
119-
actorSystem.awaitTermination()
120-
mbeanServer.unregisterMBean(siriusInfoObjectName)
121-
})
122-
123-
impl
124-
}
125-
126-
private def createSiriusInfoMBean(actorSystem: ActorSystem, siriusSup: ActorRef)
127-
(siriusConfig: SiriusConfiguration): (ObjectName, SiriusInfo) = {
128-
val resolver = siriusConfig.getProp[AkkaExternalAddressResolver](SiriusConfiguration.AKKA_EXTERNAL_ADDRESS_RESOLVER).
129-
getOrElse(throw new IllegalStateException("SiriusConfiguration.AKKA_EXTERNAL_ADDRESS_RESOLVER returned nothing"))
130-
val siriusInfo = new SiriusInfo(actorSystem, siriusSup, resolver)
131-
val objectNameHelper = new ObjectNameHelper
132-
val siriusInfoObjectName = objectNameHelper.getObjectName(siriusInfo, siriusSup, actorSystem)(siriusConfig)
133-
(siriusInfoObjectName, siriusInfo)
134-
}
135-
136-
/**
137-
* Creates configuration for the ActorSystem. The config precedence is as follows:
138-
* 1) host/port config trump all
139-
* 2) siriusConfig supplied external config next
140-
* 3) sirius-akka-base.conf, packaged with sirius, loaded with ConfigFactory.load
141-
*/
142-
private def createActorSystemConfig(siriusConfig: SiriusConfiguration): Config = {
143-
val hostPortConfig = createHostPortConfig(siriusConfig)
144-
val externalConfig = createExternalConfig(siriusConfig)
145-
val baseAkkaConfig = ConfigFactory.load("sirius-akka-base.conf")
146-
147-
hostPortConfig.withFallback(externalConfig).withFallback(baseAkkaConfig)
148-
}
149-
150-
private def createHostPortConfig(siriusConfig: SiriusConfiguration): Config = {
151-
val configMap = new JHashMap[String, Any]()
152-
val sslEnabled = siriusConfig.getProp(SiriusConfiguration.ENABLE_SSL, false)
153-
val transportPrefix = if (sslEnabled) "akka.remote.netty.ssl" else "akka.remote.netty.tcp"
154-
traceLog.info(s"AKKA using transport: $transportPrefix")
155-
156-
configMap.put("akka.remote.enabled-transports", List(transportPrefix).asJava)
157-
configMap.put(s"$transportPrefix.hostname",
158-
siriusConfig.getProp(SiriusConfiguration.HOST, InetAddress.getLocalHost.getHostName))
159-
configMap.put(s"$transportPrefix.port", siriusConfig.getProp(SiriusConfiguration.PORT, 2552))
160-
161-
val maxMessageSize = siriusConfig.getProp(SiriusConfiguration.MAX_AKKA_MESSAGE_SIZE_KB, "1024")
162-
val bufferSize = maxMessageSize * 2
163-
configMap.put(s"$transportPrefix.maximum-frame-size", s"${maxMessageSize}k")
164-
configMap.put(s"$transportPrefix.send-buffer-size", s"${bufferSize}k")
165-
configMap.put(s"$transportPrefix.receive-buffer-size", s"${bufferSize}k")
166-
167-
if (sslEnabled) {
168-
configMap.put(s"$transportPrefix.random-number-generator",
169-
siriusConfig.getProp(SiriusConfiguration.SSL_RANDOM_NUMBER_GENERATOR, ""))
170-
171-
configMap.put(s"$transportPrefix.security.key-store",
172-
siriusConfig.getProp(SiriusConfiguration.KEY_STORE_LOCATION,
173-
throw new IllegalArgumentException("No key-store provided")))
174-
175-
configMap.put(s"$transportPrefix.security.trust-store",
176-
siriusConfig.getProp(SiriusConfiguration.TRUST_STORE_LOCATION,
177-
throw new IllegalArgumentException("No trust-store provided")))
178-
179-
configMap.put(s"$transportPrefix.security.key-store-password",
180-
siriusConfig.getProp(SiriusConfiguration.KEY_STORE_PASSWORD,
181-
throw new IllegalArgumentException("No key-store-password value provided")))
182-
183-
configMap.put(s"$transportPrefix.security.key-password",
184-
siriusConfig.getProp(SiriusConfiguration.KEY_PASSWORD,
185-
throw new IllegalArgumentException("No key-password value provided")))
186-
187-
configMap.put(s"$transportPrefix.security.trust-store-password",
188-
siriusConfig.getProp(SiriusConfiguration.TRUST_STORE_PASSWORD,
189-
throw new IllegalArgumentException("No trust-store-password value provided")))
190-
}
191-
192-
// this is just so that the intellij shuts up
193-
ConfigFactory.parseMap(configMap.asInstanceOf[JHashMap[String, _ <: AnyRef]])
80+
NewSiriusFactory.createInstance(requestHandler, siriusConfig, siriusLog).asInstanceOf[SiriusImpl]
19481
}
19582

196-
/**
197-
* If siriusConfig is such configured, will load up an external configuration
198-
* for the Akka ActorSystem which is created. The filesystem is checked first,
199-
* then the classpath, if neither exist, or siriusConfig is not configured as
200-
* much, then an empty Config object is returned.
201-
*/
202-
private def createExternalConfig(siriusConfig: SiriusConfiguration): Config =
203-
siriusConfig.getProp[String](SiriusConfiguration.AKKA_EXTERN_CONFIG) match {
204-
case None => ConfigFactory.empty()
205-
case Some(externConfig) =>
206-
val externConfigFile = new File(externConfig)
207-
if (externConfigFile.exists()) {
208-
ConfigFactory.parseFile(externConfigFile).resolve()
209-
} else {
210-
ConfigFactory.parseResources(externConfig).resolve()
211-
}
212-
}
21383
}

0 commit comments

Comments
 (0)