mirror of
https://github.com/corda/corda.git
synced 2025-01-18 02:39:51 +00:00
Nodes can start without web server. Webserver now uses full node configuration.
This commit is contained in:
Clinton Alexander
parent
fa257738e1
commit
ecfb762143
@ -82,6 +82,11 @@ interface CordaRPCOps : RPCOps {
|
||||
*/
|
||||
fun nodeIdentity(): NodeInfo
|
||||
|
||||
/**
|
||||
* Returns true if node is up and ready
|
||||
*/
|
||||
fun ready(): Boolean
|
||||
|
||||
/*
|
||||
* Add note(s) to an existing Vault transaction
|
||||
*/
|
||||
|
||||
@ -1,153 +0,0 @@
|
||||
package net.corda.node.api
|
||||
|
||||
import net.corda.core.contracts.*
|
||||
import net.corda.node.api.StatesQuery
|
||||
import net.corda.core.crypto.DigitalSignature
|
||||
import net.corda.core.crypto.SecureHash
|
||||
import net.corda.core.node.NodeInfo
|
||||
import net.corda.core.serialization.SerializedBytes
|
||||
import net.corda.core.transactions.SignedTransaction
|
||||
import net.corda.core.transactions.WireTransaction
|
||||
import java.time.Instant
|
||||
import java.time.LocalDateTime
|
||||
import javax.ws.rs.GET
|
||||
import javax.ws.rs.Path
|
||||
import javax.ws.rs.Produces
|
||||
import javax.ws.rs.core.MediaType
|
||||
import javax.ws.rs.core.Response
|
||||
|
||||
/**
|
||||
* Top level interface to external interaction with the distributed ledger.
|
||||
*
|
||||
* Wherever a list is returned by a fetchXXX method that corresponds with an input list, that output list will have optional elements
|
||||
* where a null indicates "missing" and the elements returned will be in the order corresponding with the input list.
|
||||
*
|
||||
*/
|
||||
@Path("")
|
||||
interface APIServer {
|
||||
|
||||
/**
|
||||
* Report current UTC time as understood by the platform.
|
||||
*/
|
||||
@GET
|
||||
@Path("servertime")
|
||||
@Produces(MediaType.APPLICATION_JSON)
|
||||
fun serverTime(): LocalDateTime
|
||||
|
||||
/**
|
||||
* Report whether this node is started up or not.
|
||||
*/
|
||||
@GET
|
||||
@Path("status")
|
||||
@Produces(MediaType.TEXT_PLAIN)
|
||||
fun status(): Response
|
||||
|
||||
/**
|
||||
* Report this node's configuration and identities.
|
||||
* Currently tunnels the NodeInfo as an encoding of the Kryo serialised form.
|
||||
* TODO this functionality should be available via the RPC
|
||||
*/
|
||||
@GET
|
||||
@Path("info")
|
||||
@Produces(MediaType.APPLICATION_JSON)
|
||||
fun info(): NodeInfo
|
||||
|
||||
/**
|
||||
* Query your "local" states (containing only outputs involving you) and return the hashes & indexes associated with them
|
||||
* to probably be later inflated by fetchLedgerTransactions() or fetchStates() although because immutable you can cache them
|
||||
* to avoid calling fetchLedgerTransactions() many times.
|
||||
*
|
||||
* @param query Some "where clause" like expression.
|
||||
* @return Zero or more matching States.
|
||||
*/
|
||||
fun queryStates(query: StatesQuery): List<StateRef>
|
||||
|
||||
fun fetchStates(states: List<StateRef>): Map<StateRef, TransactionState<ContractState>?>
|
||||
|
||||
/**
|
||||
* Query for immutable transactions (results can be cached indefinitely by their id/hash).
|
||||
*
|
||||
* @param txs The hashes (from [StateRef.txhash] returned from [queryStates]) you would like full transactions for.
|
||||
* @return null values indicate missing transactions from the requested list.
|
||||
*/
|
||||
fun fetchTransactions(txs: List<SecureHash>): Map<SecureHash, SignedTransaction?>
|
||||
|
||||
/**
|
||||
* TransactionBuildSteps would be invocations of contract.generateXXX() methods that all share a common TransactionBuilder
|
||||
* and a common contract type (e.g. Cash or CommercialPaper)
|
||||
* which would automatically be passed as the first argument (we'd need that to be a criteria/pattern of the generateXXX methods).
|
||||
*/
|
||||
fun buildTransaction(type: ContractDefRef, steps: List<TransactionBuildStep>): SerializedBytes<WireTransaction>
|
||||
|
||||
/**
|
||||
* Generate a signature for this transaction signed by us.
|
||||
*/
|
||||
fun generateTransactionSignature(tx: SerializedBytes<WireTransaction>): DigitalSignature.WithKey
|
||||
|
||||
/**
|
||||
* Attempt to commit transaction (returned from build transaction) with the necessary signatures for that to be
|
||||
* successful, otherwise exception is thrown.
|
||||
*/
|
||||
fun commitTransaction(tx: SerializedBytes<WireTransaction>, signatures: List<DigitalSignature.WithKey>): SecureHash
|
||||
|
||||
/**
|
||||
* This method would not return until the flow is finished (hence the "Sync").
|
||||
*
|
||||
* Longer term we'd add an Async version that returns some kind of FlowInvocationRef that could be queried and
|
||||
* would appear on some kind of event message that is broadcast informing of progress.
|
||||
*
|
||||
* Will throw exception if flow fails.
|
||||
*/
|
||||
fun invokeFlowSync(type: FlowRef, args: Map<String, Any?>): Any?
|
||||
|
||||
// fun invokeFlowAsync(type: FlowRef, args: Map<String, Any?>): FlowInstanceRef
|
||||
|
||||
/**
|
||||
* Fetch flows that require a response to some prompt/question by a human (on the "bank" side).
|
||||
*/
|
||||
fun fetchFlowsRequiringAttention(query: StatesQuery): Map<StateRef, FlowRequiringAttention>
|
||||
|
||||
/**
|
||||
* Provide the response that a flow is waiting for.
|
||||
*
|
||||
* @param flow Should refer to a previously supplied FlowRequiringAttention.
|
||||
* @param stepId Which step of the flow are we referring too.
|
||||
* @param choice Should be one of the choices presented in the FlowRequiringAttention.
|
||||
* @param args Any arguments required.
|
||||
*/
|
||||
fun provideFlowResponse(flow: FlowInstanceRef, choice: SecureHash, args: Map<String, Any?>)
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
* Encapsulates the contract type. e.g. Cash or CommercialPaper etc.
|
||||
*/
|
||||
interface ContractDefRef {
|
||||
|
||||
}
|
||||
|
||||
data class ContractClassRef(val className: String) : ContractDefRef
|
||||
data class ContractLedgerRef(val hash: SecureHash) : ContractDefRef
|
||||
|
||||
|
||||
/**
|
||||
* Encapsulates the flow to be instantiated. e.g. TwoPartyTradeFlow.Buyer.
|
||||
*/
|
||||
interface FlowRef {
|
||||
|
||||
}
|
||||
|
||||
data class FlowClassRef(val className: String) : FlowRef
|
||||
|
||||
data class FlowInstanceRef(val flowInstance: SecureHash, val flowClass: FlowClassRef, val flowStepId: String)
|
||||
|
||||
/**
|
||||
* Thinking that Instant is OK for short lived flow deadlines.
|
||||
*/
|
||||
data class FlowRequiringAttention(val ref: FlowInstanceRef, val prompt: String, val choiceIdsToMessages: Map<SecureHash, String>, val dueBy: Instant)
|
||||
|
||||
|
||||
/**
|
||||
* Encapsulate a generateXXX method call on a contract.
|
||||
*/
|
||||
data class TransactionBuildStep(val generateMethodName: String, val args: Map<String, Any?>)
|
||||
@ -12,6 +12,7 @@ import com.typesafe.config.Config
|
||||
import com.typesafe.config.ConfigRenderOptions
|
||||
import net.corda.core.*
|
||||
import net.corda.core.crypto.Party
|
||||
import net.corda.core.messaging.CordaRPCOps
|
||||
import net.corda.core.node.NodeInfo
|
||||
import net.corda.core.node.services.ServiceInfo
|
||||
import net.corda.core.node.services.ServiceType
|
||||
@ -19,6 +20,9 @@ import net.corda.core.utilities.loggerFor
|
||||
import net.corda.node.services.User
|
||||
import net.corda.node.services.config.ConfigHelper
|
||||
import net.corda.node.services.config.FullNodeConfiguration
|
||||
import net.corda.node.services.config.NodeSSLConfiguration
|
||||
import net.corda.node.services.messaging.ArtemisMessagingComponent
|
||||
import net.corda.node.services.messaging.ArtemisMessagingServer
|
||||
import net.corda.node.services.messaging.CordaRPCClient
|
||||
import net.corda.node.services.messaging.NodeMessagingClient
|
||||
import net.corda.node.services.network.NetworkMapService
|
||||
@ -318,23 +322,14 @@ open class DriverDSL(
|
||||
executorService.shutdown()
|
||||
}
|
||||
|
||||
private fun queryNodeInfo(webAddress: HostAndPort): NodeInfo? {
|
||||
val url = URL("http://$webAddress/api/info")
|
||||
private fun queryNodeInfo(webAddress: HostAndPort, sslConfig: NodeSSLConfiguration): NodeInfo? {
|
||||
try {
|
||||
val conn = url.openConnection() as HttpURLConnection
|
||||
conn.requestMethod = "GET"
|
||||
if (conn.responseCode != 200) {
|
||||
log.error("Received response code ${conn.responseCode} from $url during startup.")
|
||||
return null
|
||||
}
|
||||
// For now the NodeInfo is tunneled in its Kryo format over the Node's Web interface.
|
||||
val om = ObjectMapper()
|
||||
val module = SimpleModule("NodeInfo")
|
||||
module.addDeserializer(NodeInfo::class.java, JsonSupport.NodeInfoDeserializer)
|
||||
om.registerModule(module)
|
||||
return om.readValue(conn.inputStream, NodeInfo::class.java)
|
||||
val client = CordaRPCClient(webAddress, sslConfig)
|
||||
client.start(ArtemisMessagingComponent.NODE_USER, ArtemisMessagingComponent.NODE_USER)
|
||||
val rpcOps = client.proxy()
|
||||
return rpcOps.nodeIdentity()
|
||||
} catch(e: Exception) {
|
||||
log.error("Could not query node info at $url due to an exception.", e)
|
||||
log.error("Could not query node info at $webAddress due to an exception.", e)
|
||||
return null
|
||||
}
|
||||
}
|
||||
@ -378,7 +373,7 @@ open class DriverDSL(
|
||||
val startNode = startNode(executorService, configuration, quasarJarPath, debugPort)
|
||||
registerProcess(startNode)
|
||||
return startNode.map {
|
||||
NodeHandle(queryNodeInfo(apiAddress)!!, configuration, it)
|
||||
NodeHandle(queryNodeInfo(messagingAddress, configuration)!!, configuration, it)
|
||||
}
|
||||
}
|
||||
|
||||
@ -488,9 +483,7 @@ open class DriverDSL(
|
||||
return Futures.allAsList(
|
||||
addressMustBeBound(executorService, nodeConf.artemisAddress),
|
||||
// TODO There is a race condition here. Even though the messaging address is bound it may be the case that
|
||||
// the handlers for the advertised services are not yet registered. A hacky workaround is that we wait for
|
||||
// the web api address to be bound as well, as that starts after the services. Needs rethinking.
|
||||
addressMustBeBound(executorService, nodeConf.webAddress)
|
||||
// the handlers for the advertised services are not yet registered. Needs rethinking.
|
||||
).map { process }
|
||||
}
|
||||
}
|
||||
|
||||
@ -1,92 +0,0 @@
|
||||
package net.corda.node.internal
|
||||
|
||||
import com.google.common.util.concurrent.ListenableFuture
|
||||
import net.corda.core.contracts.ContractState
|
||||
import net.corda.core.contracts.DealState
|
||||
import net.corda.core.contracts.StateRef
|
||||
import net.corda.core.contracts.TransactionState
|
||||
import net.corda.core.crypto.DigitalSignature
|
||||
import net.corda.core.crypto.SecureHash
|
||||
import net.corda.core.node.services.linearHeadsOfType
|
||||
import net.corda.core.serialization.SerializedBytes
|
||||
import net.corda.core.transactions.SignedTransaction
|
||||
import net.corda.core.transactions.WireTransaction
|
||||
import net.corda.node.api.*
|
||||
import java.time.LocalDateTime
|
||||
import javax.ws.rs.core.Response
|
||||
|
||||
class APIServerImpl(val node: AbstractNode) : APIServer {
|
||||
|
||||
override fun serverTime(): LocalDateTime = LocalDateTime.now(node.services.clock)
|
||||
|
||||
override fun status(): Response {
|
||||
return if (node.started) {
|
||||
Response.ok("started").build()
|
||||
} else {
|
||||
Response.status(Response.Status.SERVICE_UNAVAILABLE).entity("not started").build()
|
||||
}
|
||||
}
|
||||
|
||||
override fun info() = node.services.myInfo
|
||||
|
||||
override fun queryStates(query: StatesQuery): List<StateRef> {
|
||||
// We're going to hard code two options here for now and assume that all LinearStates are deals
|
||||
// Would like to maybe move to a model where we take something like a JEXL string, although don't want to develop
|
||||
// something we can't later implement against a persistent store (i.e. need to pick / build a query engine)
|
||||
if (query is StatesQuery.Selection) {
|
||||
if (query.criteria is StatesQuery.Criteria.AllDeals) {
|
||||
val states = node.services.vaultService.linearHeads
|
||||
return states.values.map { it.ref }
|
||||
} else if (query.criteria is StatesQuery.Criteria.Deal) {
|
||||
val states = node.services.vaultService.linearHeadsOfType<DealState>().filterValues {
|
||||
it.state.data.ref == query.criteria.ref
|
||||
}
|
||||
return states.values.map { it.ref }
|
||||
}
|
||||
}
|
||||
return emptyList()
|
||||
}
|
||||
|
||||
override fun fetchStates(states: List<StateRef>): Map<StateRef, TransactionState<ContractState>?> {
|
||||
return node.services.vaultService.statesForRefs(states)
|
||||
}
|
||||
|
||||
override fun fetchTransactions(txs: List<SecureHash>): Map<SecureHash, SignedTransaction?> {
|
||||
throw UnsupportedOperationException()
|
||||
}
|
||||
|
||||
override fun buildTransaction(type: ContractDefRef, steps: List<TransactionBuildStep>): SerializedBytes<WireTransaction> {
|
||||
throw UnsupportedOperationException()
|
||||
}
|
||||
|
||||
override fun generateTransactionSignature(tx: SerializedBytes<WireTransaction>): DigitalSignature.WithKey {
|
||||
throw UnsupportedOperationException()
|
||||
}
|
||||
|
||||
override fun commitTransaction(tx: SerializedBytes<WireTransaction>, signatures: List<DigitalSignature.WithKey>): SecureHash {
|
||||
throw UnsupportedOperationException()
|
||||
}
|
||||
|
||||
override fun invokeFlowSync(type: FlowRef, args: Map<String, Any?>): Any? {
|
||||
return invokeFlowAsync(type, args).get()
|
||||
}
|
||||
|
||||
private fun invokeFlowAsync(type: FlowRef, args: Map<String, Any?>): ListenableFuture<out Any?> {
|
||||
if (type is FlowClassRef) {
|
||||
val flowLogicRef = node.services.flowLogicRefFactory.createKotlin(type.className, args)
|
||||
val flowInstance = node.services.flowLogicRefFactory.toFlowLogic(flowLogicRef)
|
||||
return node.services.startFlow(flowInstance).resultFuture
|
||||
} else {
|
||||
throw UnsupportedOperationException("Unsupported FlowRef type: $type")
|
||||
}
|
||||
}
|
||||
|
||||
override fun fetchFlowsRequiringAttention(query: StatesQuery): Map<StateRef, FlowRequiringAttention> {
|
||||
throw UnsupportedOperationException()
|
||||
}
|
||||
|
||||
override fun provideFlowResponse(flow: FlowInstanceRef, choice: SecureHash, args: Map<String, Any?>) {
|
||||
throw UnsupportedOperationException()
|
||||
}
|
||||
|
||||
}
|
||||
@ -24,7 +24,6 @@ import net.corda.flows.CashCommand
|
||||
import net.corda.flows.CashFlow
|
||||
import net.corda.flows.FinalityFlow
|
||||
import net.corda.flows.sendRequest
|
||||
import net.corda.node.api.APIServer
|
||||
import net.corda.node.services.api.*
|
||||
import net.corda.node.services.config.NodeConfiguration
|
||||
import net.corda.node.services.config.configureWithDevSSLCertificate
|
||||
@ -163,7 +162,6 @@ abstract class AbstractNode(open val configuration: NodeConfiguration,
|
||||
lateinit var identity: IdentityService
|
||||
lateinit var net: MessagingServiceInternal
|
||||
lateinit var netMapCache: NetworkMapCache
|
||||
lateinit var api: APIServer
|
||||
lateinit var scheduler: NodeSchedulerService
|
||||
lateinit var flowLogicFactory: FlowLogicRefFactory
|
||||
lateinit var schemas: SchemaService
|
||||
@ -219,7 +217,6 @@ abstract class AbstractNode(open val configuration: NodeConfiguration,
|
||||
// the KMS is meant for derived temporary keys used in transactions, and we're not supposed to sign things with
|
||||
// the identity key. But the infrastructure to make that easy isn't here yet.
|
||||
keyManagement = makeKeyManagementService()
|
||||
api = APIServerImpl(this@AbstractNode)
|
||||
flowLogicFactory = initialiseFlowLogicFactory()
|
||||
scheduler = NodeSchedulerService(database, services, flowLogicFactory, unfinishedSchedules = busyNodeLatch)
|
||||
|
||||
|
||||
@ -78,6 +78,10 @@ class CordaRPCOpsImpl(
|
||||
return services.myInfo
|
||||
}
|
||||
|
||||
override fun ready(): Boolean {
|
||||
return true
|
||||
}
|
||||
|
||||
override fun addVaultTransactionNote(txnId: SecureHash, txnNote: String) {
|
||||
return databaseTransaction(database) {
|
||||
services.vaultService.addNoteToTransaction(txnId, txnNote)
|
||||
|
||||
@ -111,7 +111,6 @@ class Node(override val configuration: FullNodeConfiguration,
|
||||
// serialisation/deserialisation work.
|
||||
override val serverThread = AffinityExecutor.ServiceAffinityExecutor("Node thread", 1)
|
||||
|
||||
//lateinit var webServer: Server
|
||||
var messageBroker: ArtemisMessagingServer? = null
|
||||
|
||||
// Avoid the lock being garbage collected. We don't really need to release it as the OS will do so for us
|
||||
@ -157,116 +156,6 @@ class Node(override val configuration: FullNodeConfiguration,
|
||||
return networkMapConnection.flatMap { super.registerWithNetworkMap() }
|
||||
}
|
||||
|
||||
// TODO: add flag to enable/disable webserver
|
||||
private fun initWebServer(localRpc: CordaRPCOps): Server {
|
||||
// Note that the web server handlers will all run concurrently, and not on the node thread.
|
||||
val handlerCollection = HandlerCollection()
|
||||
|
||||
// Export JMX monitoring statistics and data over REST/JSON.
|
||||
if (configuration.exportJMXto.split(',').contains("http")) {
|
||||
val classpath = System.getProperty("java.class.path").split(System.getProperty("path.separator"))
|
||||
val warpath = classpath.firstOrNull { it.contains("jolokia-agent-war-2") && it.endsWith(".war") }
|
||||
if (warpath != null) {
|
||||
handlerCollection.addHandler(WebAppContext().apply {
|
||||
// Find the jolokia WAR file on the classpath.
|
||||
contextPath = "/monitoring/json"
|
||||
setInitParameter("mimeType", "application/json")
|
||||
war = warpath
|
||||
})
|
||||
} else {
|
||||
log.warn("Unable to locate Jolokia WAR on classpath")
|
||||
}
|
||||
}
|
||||
|
||||
// API, data upload and download to services (attachments, rates oracles etc)
|
||||
handlerCollection.addHandler(buildServletContextHandler(localRpc))
|
||||
|
||||
val server = Server()
|
||||
|
||||
val connector = if (configuration.useHTTPS) {
|
||||
val httpsConfiguration = HttpConfiguration()
|
||||
httpsConfiguration.outputBufferSize = 32768
|
||||
httpsConfiguration.addCustomizer(SecureRequestCustomizer())
|
||||
val sslContextFactory = SslContextFactory()
|
||||
sslContextFactory.keyStorePath = configuration.keyStoreFile.toString()
|
||||
sslContextFactory.setKeyStorePassword(configuration.keyStorePassword)
|
||||
sslContextFactory.setKeyManagerPassword(configuration.keyStorePassword)
|
||||
sslContextFactory.setTrustStorePath(configuration.trustStoreFile.toString())
|
||||
sslContextFactory.setTrustStorePassword(configuration.trustStorePassword)
|
||||
sslContextFactory.setExcludeProtocols("SSL.*", "TLSv1", "TLSv1.1")
|
||||
sslContextFactory.setIncludeProtocols("TLSv1.2")
|
||||
sslContextFactory.setExcludeCipherSuites(".*NULL.*", ".*RC4.*", ".*MD5.*", ".*DES.*", ".*DSS.*")
|
||||
sslContextFactory.setIncludeCipherSuites(".*AES.*GCM.*")
|
||||
val sslConnector = ServerConnector(server, SslConnectionFactory(sslContextFactory, "http/1.1"), HttpConnectionFactory(httpsConfiguration))
|
||||
sslConnector.port = configuration.webAddress.port
|
||||
sslConnector
|
||||
} else {
|
||||
val httpConfiguration = HttpConfiguration()
|
||||
httpConfiguration.outputBufferSize = 32768
|
||||
val httpConnector = ServerConnector(server, HttpConnectionFactory(httpConfiguration))
|
||||
httpConnector.port = configuration.webAddress.port
|
||||
httpConnector
|
||||
}
|
||||
server.connectors = arrayOf<Connector>(connector)
|
||||
|
||||
server.handler = handlerCollection
|
||||
runOnStop += Runnable { server.stop() }
|
||||
server.start()
|
||||
printBasicNodeInfo("Embedded web server is listening on", "http://${InetAddress.getLocalHost().hostAddress}:${connector.port}/")
|
||||
return server
|
||||
}
|
||||
|
||||
private fun buildServletContextHandler(localRpc: CordaRPCOps): ServletContextHandler {
|
||||
return ServletContextHandler().apply {
|
||||
contextPath = "/"
|
||||
setAttribute("node", this@Node)
|
||||
addServlet(DataUploadServlet::class.java, "/upload/*")
|
||||
addServlet(AttachmentDownloadServlet::class.java, "/attachments/*")
|
||||
|
||||
val resourceConfig = ResourceConfig()
|
||||
// Add your API provider classes (annotated for JAX-RS) here
|
||||
resourceConfig.register(Config(services))
|
||||
resourceConfig.register(ResponseFilter())
|
||||
resourceConfig.register(api)
|
||||
|
||||
val webAPIsOnClasspath = pluginRegistries.flatMap { x -> x.webApis }
|
||||
for (webapi in webAPIsOnClasspath) {
|
||||
log.info("Add plugin web API from attachment $webapi")
|
||||
val customAPI = try {
|
||||
webapi.apply(localRpc)
|
||||
} catch (ex: InvocationTargetException) {
|
||||
log.error("Constructor $webapi threw an error: ", ex.targetException)
|
||||
continue
|
||||
}
|
||||
resourceConfig.register(customAPI)
|
||||
}
|
||||
|
||||
val staticDirMaps = pluginRegistries.map { x -> x.staticServeDirs }
|
||||
val staticDirs = staticDirMaps.flatMap { it.keys }.zip(staticDirMaps.flatMap { it.values })
|
||||
staticDirs.forEach {
|
||||
val staticDir = ServletHolder(DefaultServlet::class.java)
|
||||
staticDir.setInitParameter("resourceBase", it.second)
|
||||
staticDir.setInitParameter("dirAllowed", "true")
|
||||
staticDir.setInitParameter("pathInfoOnly", "true")
|
||||
addServlet(staticDir, "/web/${it.first}/*")
|
||||
}
|
||||
|
||||
// Give the app a slightly better name in JMX rather than a randomly generated one and enable JMX
|
||||
resourceConfig.addProperties(mapOf(ServerProperties.APPLICATION_NAME to "node.api",
|
||||
ServerProperties.MONITORING_STATISTICS_MBEANS_ENABLED to "true"))
|
||||
|
||||
val container = ServletContainer(resourceConfig)
|
||||
val jerseyServlet = ServletHolder(container)
|
||||
addServlet(jerseyServlet, "/api/*")
|
||||
jerseyServlet.initOrder = 0 // Initialise at server start
|
||||
|
||||
// Wrap all API calls in a database transaction.
|
||||
val filterHolder = FilterHolder(DatabaseTransactionFilter(database))
|
||||
addFilter(filterHolder, "/api/*", EnumSet.of(DispatcherType.REQUEST))
|
||||
addFilter(filterHolder, "/upload/*", EnumSet.of(DispatcherType.REQUEST))
|
||||
}
|
||||
}
|
||||
|
||||
override fun makeUniquenessProvider(type: ServiceType): UniquenessProvider {
|
||||
return when (type) {
|
||||
RaftValidatingNotaryService.type -> with(configuration) {
|
||||
@ -305,26 +194,12 @@ class Node(override val configuration: FullNodeConfiguration,
|
||||
super.initialiseDatabasePersistence(insideTransaction)
|
||||
}
|
||||
|
||||
private fun connectLocalRpcAsNodeUser(): CordaRPCOps {
|
||||
val client = CordaRPCClient(configuration.artemisAddress, configuration)
|
||||
client.start(NODE_USER, NODE_USER)
|
||||
return client.proxy()
|
||||
}
|
||||
|
||||
override fun start(): Node {
|
||||
alreadyRunningNodeCheck()
|
||||
super.start()
|
||||
// Only start the service API requests once the network map registration is complete
|
||||
thread(name = "WebServer") {
|
||||
networkMapRegistrationFuture.getOrThrow()
|
||||
// TODO: Remove when cleanup
|
||||
//try {
|
||||
// webServer = initWebServer(connectLocalRpcAsNodeUser())
|
||||
//} catch(ex: Exception) {
|
||||
// // TODO: We need to decide if this is a fatal error, given the API is unavailable, or whether the API
|
||||
// // is not critical and we continue anyway.
|
||||
// log.error("Web server startup failed", ex)
|
||||
//}
|
||||
// Begin exporting our own metrics via JMX.
|
||||
JmxReporter.
|
||||
forRegistry(services.monitoringService.metrics).
|
||||
|
||||
@ -30,6 +30,11 @@ dependencies {
|
||||
compile "org.jolokia:jolokia-agent-war:2.0.0-M1"
|
||||
compile "commons-fileupload:commons-fileupload:1.3.2"
|
||||
|
||||
// Log4J: logging framework (with SLF4J bindings)
|
||||
compile "org.apache.logging.log4j:log4j-slf4j-impl:${log4j_version}"
|
||||
compile "org.apache.logging.log4j:log4j-core:${log4j_version}"
|
||||
compile "org.apache.logging.log4j:log4j-web:${log4j_version}"
|
||||
|
||||
// Jersey for JAX-RS implementation for use in Jetty
|
||||
compile "org.glassfish.jersey.core:jersey-server:${jersey_version}"
|
||||
compile "org.glassfish.jersey.containers:jersey-container-servlet-core:${jersey_version}"
|
||||
|
||||
@ -1,48 +1,14 @@
|
||||
package net.corda.node.webserver
|
||||
|
||||
import com.google.common.net.HostAndPort
|
||||
import net.corda.node.driver.driver
|
||||
import net.corda.node.services.User
|
||||
import net.corda.node.services.config.ConfigHelper
|
||||
import net.corda.node.services.config.FullNodeConfiguration
|
||||
import java.nio.file.Paths
|
||||
|
||||
fun main(args: Array<String>) {
|
||||
// TODO: Print basic webserver info
|
||||
System.setProperty("consoleLogLevel", "info")
|
||||
driver {
|
||||
val node = startNode().get()
|
||||
WebServer(node.nodeInfo, node.config).start()
|
||||
val server = WebServer(FullNodeConfiguration(node.config)).start()
|
||||
waitForAllNodesToFinish()
|
||||
}
|
||||
}
|
||||
|
||||
fun generateNodeConfiguration(): FullNodeConfiguration {
|
||||
val messagingAddress = 10002
|
||||
val apiAddress = HostAndPort.fromString("localhost:10003")
|
||||
val name = "webserver-test"
|
||||
val rpcUsers = listOf<User>()
|
||||
|
||||
val baseDirectory = Paths.get("build/webserver")
|
||||
val configOverrides = mapOf(
|
||||
"myLegalName" to name,
|
||||
"basedir" to baseDirectory.normalize().toString(),
|
||||
"artemisAddress" to messagingAddress.toString(),
|
||||
"webAddress" to apiAddress.toString(),
|
||||
"extraAdvertisedServiceIds" to listOf<String>(),
|
||||
"networkMapAddress" to "",
|
||||
"useTestClock" to false,
|
||||
"rpcUsers" to rpcUsers.map {
|
||||
mapOf(
|
||||
"user" to it.username,
|
||||
"password" to it.password,
|
||||
"permissions" to it.permissions
|
||||
)
|
||||
}
|
||||
)
|
||||
|
||||
val config = ConfigHelper.loadConfig(
|
||||
baseDirectoryPath = baseDirectory,
|
||||
allowMissingConfig = true,
|
||||
configOverrides = configOverrides
|
||||
)
|
||||
|
||||
return FullNodeConfiguration(config)
|
||||
}
|
||||
@ -1,23 +1,18 @@
|
||||
package net.corda.node.webserver
|
||||
|
||||
import com.google.common.net.HostAndPort
|
||||
import com.typesafe.config.Config
|
||||
import net.corda.core.messaging.CordaRPCOps
|
||||
import net.corda.core.node.CordaPluginRegistry
|
||||
import net.corda.core.node.NodeInfo
|
||||
import net.corda.core.utilities.loggerFor
|
||||
import net.corda.node.internal.Node
|
||||
import net.corda.node.services.config.FullNodeConfiguration
|
||||
import net.corda.node.services.config.NodeSSLConfiguration
|
||||
import net.corda.node.services.messaging.ArtemisMessagingComponent
|
||||
import net.corda.node.services.messaging.CordaRPCClient
|
||||
import net.corda.node.servlets.AttachmentDownloadServlet
|
||||
import net.corda.node.servlets.DataUploadServlet
|
||||
import net.corda.node.servlets.ResponseFilter
|
||||
import net.corda.node.webserver.internal.APIServerImpl
|
||||
import org.eclipse.jetty.server.*
|
||||
import org.eclipse.jetty.server.handler.HandlerCollection
|
||||
import org.eclipse.jetty.servlet.DefaultServlet
|
||||
import org.eclipse.jetty.servlet.FilterHolder
|
||||
import org.eclipse.jetty.servlet.ServletContextHandler
|
||||
import org.eclipse.jetty.servlet.ServletHolder
|
||||
import org.eclipse.jetty.util.ssl.SslContextFactory
|
||||
@ -27,25 +22,14 @@ import org.glassfish.jersey.server.ServerProperties
|
||||
import org.glassfish.jersey.servlet.ServletContainer
|
||||
import java.lang.reflect.InvocationTargetException
|
||||
import java.net.InetAddress
|
||||
import java.nio.file.Path
|
||||
import java.util.*
|
||||
import javax.servlet.DispatcherType
|
||||
|
||||
class WebServer(val nodeInfo: NodeInfo, val configuration: Config) {
|
||||
class WebServer(val config: FullNodeConfiguration) {
|
||||
private companion object {
|
||||
val log = loggerFor<WebServer>()
|
||||
}
|
||||
|
||||
private val address = HostAndPort.fromString(configuration.getString("webAddress"))
|
||||
private val sslConfig = object : NodeSSLConfiguration {
|
||||
override val keyStorePassword: String
|
||||
get() = throw UnsupportedOperationException()
|
||||
override val trustStorePassword: String
|
||||
get() = throw UnsupportedOperationException()
|
||||
override val certificatesPath: Path
|
||||
get() = throw UnsupportedOperationException()
|
||||
|
||||
}
|
||||
val address = config.webAddress
|
||||
|
||||
fun start() {
|
||||
initWebServer(connectLocalRpcAsNodeUser())
|
||||
@ -56,7 +40,7 @@ class WebServer(val nodeInfo: NodeInfo, val configuration: Config) {
|
||||
val handlerCollection = HandlerCollection()
|
||||
|
||||
// Export JMX monitoring statistics and data over REST/JSON.
|
||||
if (configuration.getString("exportJMXto").split(',').contains("http")) {
|
||||
if (config.exportJMXto.split(',').contains("http")) {
|
||||
val classpath = System.getProperty("java.class.path").split(System.getProperty("path.separator"))
|
||||
val warpath = classpath.firstOrNull { it.contains("jolokia-agent-war-2") && it.endsWith(".war") }
|
||||
if (warpath != null) {
|
||||
@ -76,16 +60,16 @@ class WebServer(val nodeInfo: NodeInfo, val configuration: Config) {
|
||||
|
||||
val server = Server()
|
||||
|
||||
val connector = if (configuration.getBoolean("useHTTPS")) {
|
||||
val connector = if (config.useHTTPS) {
|
||||
val httpsConfiguration = HttpConfiguration()
|
||||
httpsConfiguration.outputBufferSize = 32768
|
||||
httpsConfiguration.addCustomizer(SecureRequestCustomizer())
|
||||
val sslContextFactory = SslContextFactory()
|
||||
sslContextFactory.keyStorePath = sslConfig.keyStorePath.toString()
|
||||
sslContextFactory.setKeyStorePassword(sslConfig.keyStorePassword)
|
||||
sslContextFactory.setKeyManagerPassword(sslConfig.keyStorePassword)
|
||||
sslContextFactory.setTrustStorePath(sslConfig.trustStorePath.toString())
|
||||
sslContextFactory.setTrustStorePassword(sslConfig.trustStorePassword)
|
||||
sslContextFactory.keyStorePath = config.keyStorePath.toString()
|
||||
sslContextFactory.setKeyStorePassword(config.keyStorePassword)
|
||||
sslContextFactory.setKeyManagerPassword(config.keyStorePassword)
|
||||
sslContextFactory.setTrustStorePath(config.trustStorePath.toString())
|
||||
sslContextFactory.setTrustStorePassword(config.trustStorePassword)
|
||||
sslContextFactory.setExcludeProtocols("SSL.*", "TLSv1", "TLSv1.1")
|
||||
sslContextFactory.setIncludeProtocols("TLSv1.2")
|
||||
sslContextFactory.setExcludeCipherSuites(".*NULL.*", ".*RC4.*", ".*MD5.*", ".*DES.*", ".*DSS.*")
|
||||
@ -97,6 +81,7 @@ class WebServer(val nodeInfo: NodeInfo, val configuration: Config) {
|
||||
val httpConfiguration = HttpConfiguration()
|
||||
httpConfiguration.outputBufferSize = 32768
|
||||
val httpConnector = ServerConnector(server, HttpConnectionFactory(httpConfiguration))
|
||||
println("Starting webserver on address $address")
|
||||
httpConnector.port = address.port
|
||||
httpConnector
|
||||
}
|
||||
@ -112,19 +97,16 @@ class WebServer(val nodeInfo: NodeInfo, val configuration: Config) {
|
||||
private fun buildServletContextHandler(localRpc: CordaRPCOps): ServletContextHandler {
|
||||
return ServletContextHandler().apply {
|
||||
contextPath = "/"
|
||||
//setAttribute("node", this@Node)
|
||||
addServlet(DataUploadServlet::class.java, "/upload/*")
|
||||
addServlet(AttachmentDownloadServlet::class.java, "/attachments/*")
|
||||
|
||||
val resourceConfig = ResourceConfig()
|
||||
// Add your API provider classes (annotated for JAX-RS) here
|
||||
// TODO: Remove this at cleanup time
|
||||
//resourceConfig.register(Config(services))
|
||||
resourceConfig.register(ResponseFilter())
|
||||
// TODO: Move the API out of node and to here.
|
||||
//resourceConfig.register(api)
|
||||
resourceConfig.register(APIServerImpl(localRpc))
|
||||
|
||||
val webAPIsOnClasspath = pluginRegistries.flatMap { x -> x.webApis }
|
||||
println("NUM PLUGINS: ${pluginRegistries.size}")
|
||||
println("NUM WEBAPIS: ${webAPIsOnClasspath.size}")
|
||||
for (webapi in webAPIsOnClasspath) {
|
||||
log.info("Add plugin web API from attachment $webapi")
|
||||
val customAPI = try {
|
||||
@ -154,17 +136,12 @@ class WebServer(val nodeInfo: NodeInfo, val configuration: Config) {
|
||||
val jerseyServlet = ServletHolder(container)
|
||||
addServlet(jerseyServlet, "/api/*")
|
||||
jerseyServlet.initOrder = 0 // Initialise at server start
|
||||
|
||||
// Wrap all API calls in a database transaction.
|
||||
// TODO: Remove this when cleaning up
|
||||
//val filterHolder = FilterHolder(Node.DatabaseTransactionFilter(database))
|
||||
//addFilter(filterHolder, "/api/*", EnumSet.of(DispatcherType.REQUEST))
|
||||
//addFilter(filterHolder, "/upload/*", EnumSet.of(DispatcherType.REQUEST))
|
||||
}
|
||||
}
|
||||
|
||||
private fun connectLocalRpcAsNodeUser(): CordaRPCOps {
|
||||
val client = CordaRPCClient(HostAndPort.fromString(nodeInfo.address.toString()), sslConfig)
|
||||
log.info("Connecting to node at ${config.artemisAddress} as node user")
|
||||
val client = CordaRPCClient(config.artemisAddress, config)
|
||||
client.start(ArtemisMessagingComponent.NODE_USER, ArtemisMessagingComponent.NODE_USER)
|
||||
return client.proxy()
|
||||
}
|
||||
|
||||
@ -0,0 +1,43 @@
|
||||
package net.corda.node.webserver.api
|
||||
|
||||
import net.corda.core.node.NodeInfo
|
||||
import java.time.LocalDateTime
|
||||
import javax.ws.rs.GET
|
||||
import javax.ws.rs.Path
|
||||
import javax.ws.rs.Produces
|
||||
import javax.ws.rs.core.MediaType
|
||||
import javax.ws.rs.core.Response
|
||||
|
||||
/**
|
||||
* Top level interface to external interaction with the distributed ledger.
|
||||
*
|
||||
* Wherever a list is returned by a fetchXXX method that corresponds with an input list, that output list will have optional elements
|
||||
* where a null indicates "missing" and the elements returned will be in the order corresponding with the input list.
|
||||
*
|
||||
*/
|
||||
@Path("")
|
||||
interface APIServer {
|
||||
/**
|
||||
* Report current UTC time as understood by the platform.
|
||||
*/
|
||||
@GET
|
||||
@Path("servertime")
|
||||
@Produces(MediaType.APPLICATION_JSON)
|
||||
fun serverTime(): LocalDateTime
|
||||
|
||||
/**
|
||||
* Report whether this node is started up or not.
|
||||
*/
|
||||
@GET
|
||||
@Path("status")
|
||||
@Produces(MediaType.TEXT_PLAIN)
|
||||
fun status(): Response
|
||||
|
||||
/**
|
||||
* Report this node's configuration and identities.
|
||||
*/
|
||||
@GET
|
||||
@Path("info")
|
||||
@Produces(MediaType.APPLICATION_JSON)
|
||||
fun info(): NodeInfo
|
||||
}
|
||||
@ -1,4 +1,4 @@
|
||||
package net.corda.node.api
|
||||
package net.corda.node.webserver.api
|
||||
|
||||
/**
|
||||
* Extremely rudimentary query language which should most likely be replaced with a product.
|
||||
@ -0,0 +1,24 @@
|
||||
package net.corda.node.webserver.internal
|
||||
|
||||
import net.corda.core.messaging.CordaRPCOps
|
||||
import net.corda.node.webserver.api.*
|
||||
import java.time.LocalDateTime
|
||||
import java.time.ZoneId
|
||||
import javax.ws.rs.core.Response
|
||||
|
||||
class APIServerImpl(val rpcOps: CordaRPCOps) : APIServer {
|
||||
|
||||
override fun serverTime(): LocalDateTime {
|
||||
return LocalDateTime.ofInstant(rpcOps.currentNodeTime(), ZoneId.of("UTC"))
|
||||
}
|
||||
|
||||
override fun status(): Response {
|
||||
return if (rpcOps.ready()) {
|
||||
Response.ok("started").build()
|
||||
} else {
|
||||
Response.status(Response.Status.SERVICE_UNAVAILABLE).entity("not started").build()
|
||||
}
|
||||
}
|
||||
|
||||
override fun info() = rpcOps.nodeIdentity()
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user