test/repchain
1
0
Fork 0
mirror of https://gitee.com/BTAJL/repchain.git synced 2024-12-04 12:48:32 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

proto ChaincodeId

Browse Source
This commit is contained in:
chen4w 2019-03-11 09:26:04 +08:00
commit 3ed83eb024
25 changed files with 3858 additions and 3844 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

212
conf/logback.xml
View File

@ -1,108 +1,106 @@
<configuration DEBUG="true">
<turboFilter class="rep.log.trace.repTurboLogFilter"/>
<!--Tips: Appender configuration, more detail ~ https://logback.qos.ch/manual/appenders.html-->
<!-- Insert the current time formatted as "yyyyMMdd'T'HHmmss" under
the key "bySecond" into the logger context. This value will be
available to all subsequent configuration elements. -->
<!--<timestamp key="bySecond" datePattern="yyyyMMdd'T'HHmmss"/>-->
<!--Advised：File logging, more configurable-->
<appender name="FILE" class="ch.qos.logback.core.rolling.RollingFileAppender">
<rollingPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
<!-- rollover daily -->
<fileNamePattern>logs/mylog-%d{yyyy-MM-dd}.%i.log</fileNamePattern>
<!-- each file should be at most 100MB, keep 30 days worth of history, but at most 2GB -->
<maxFileSize>100MB</maxFileSize>
<maxHistory>30</maxHistory>
<totalSizeCap>2GB</totalSizeCap>
</rollingPolicy>
<!-- <filter class="ch.qos.logback.classic.filter.LevelFilter"> -->
<filter class="ch.qos.logback.classic.filter.ThresholdFilter">
<level>DEBUG</level>
<!--Options:ACCEPT,NEUTRAL,DENY-->
<!-- <onMatch>ACCEPT</onMatch> -->
<!-- <onMismatch>DENY</onMismatch> -->
</filter>
<immediateFlush>false</immediateFlush>
<append>false</append>
<encoder>
<pattern>%date{ISO8601} %-5level %logger{36} %X{akkaSource} - %msg%n</pattern>
<!--<pattern>%-4relative [%thread] %-5level %logger{35} - %msg%n</pattern>-->
</encoder>
</appender>
<!--For async the logging, drop the info and debug when the block is filled of 80% size by default, maxFlushTime by default-->
<!--<appender name="ASYNC" class="ch.qos.logback.classic.AsyncAppender">-->
<!--<appender-ref ref="FILE" />-->
<!--<nerverBlock>true</nerverBlock>-->
<!--</appender>-->
<!--Deprecated ：Basic file appender configure()-->
<!--<appender name="FILE" class="ch.qos.logback.core.FileAppender">-->
<!--<file>logs/repchain-${bySecond}.log</file>-->
<!--<filter class="ch.qos.logback.classic.filter.ThresholdFilter">-->
<!--<level>WARN</level>-->
<!--</filter>-->
<!--&lt;!&ndash; set immediateFlush to false for much higher logging throughput &ndash;&gt;-->
<!--<immediateFlush>false</immediateFlush>-->
<!--<append>false</append>-->
<!--&lt;!&ndash; encoders are assigned the type-->
<!--ch.qos.logback.classic.encoder.PatternLayoutEncoder by default &ndash;&gt;-->
<!--<encoder>-->
<!--&lt;!&ndash;<pattern>%date{ISO8601} %-5level %logger{36} %X{sourceThread} - %msg%n</pattern>&ndash;&gt;-->
<!--<pattern>%date{ISO8601} %-5level %logger{36} %X{akkaSource} - %msg%n</pattern>-->
<!--&lt;!&ndash;<pattern>%date{yyyy-MM-dd} %X{akkaTimestamp} %-5level[%thread] %logger{1} - %msg%n</pattern>&ndash;&gt;-->
<!--</encoder>-->
<!--</appender>-->
<appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
<target>System.out</target>
<!--level过滤器简化版-->
<filter class="ch.qos.logback.classic.filter.ThresholdFilter">
<level>DEBUG</level>
</filter>
<!--Java 过滤器-->
<!--filter class="ch.qos.logback.core.filter.EvaluatorFilter">
<evaluator>
<expression>return message.contains("Statistic");</expression>
</evaluator>
<OnMismatch>DENY</OnMismatch>
<OnMatch>ACCEPT</OnMatch>
</filter-->
<!--正则表达式过滤器-->
<!--<filter class="ch.qos.logback.core.filter.EvaluatorFilter">-->
<!--<evaluator>-->
<!--<matcher>-->
<!--<Name>pm_1M</Name>-->
<!--&lt;!&ndash; filter out odd numbered statements &ndash;&gt;-->
<!--<regex>block</regex>-->
<!--</matcher>-->
<!--<expression>pm_1M.matches(message)</expression>-->
<!--</evaluator>-->
<!--<OnMismatch>DENY</OnMismatch>-->
<!--<OnMatch>NEUTRAL</OnMatch>-->
<!--</filter>-->
<encoder>
<pattern>%date{ISO8601} %-5level %logger{36} %X{akkaSource} - %msg%n</pattern>
<!--<pattern>%X{akkaTimestamp} %-5level %logger{36} %X{akkaSource} - %msg%n</pattern>-->
<!--<pattern>%X{akkaTimestamp} %-5level[%thread] %logger{0} - %msg%n</pattern>-->
</encoder>
</appender>
<!--<logger name="akka" level="INFO" />-->
<root level="DEBUG">
<appender-ref ref="STDOUT" />
<appender-ref ref="FILE" />
</root>
<configuration DEBUG="true">
<!--Tips: Appender configuration, more detail ~ https://logback.qos.ch/manual/appenders.html-->
<!-- Insert the current time formatted as "yyyyMMdd'T'HHmmss" under
the key "bySecond" into the logger context. This value will be
available to all subsequent configuration elements. -->
<!--<timestamp key="bySecond" datePattern="yyyyMMdd'T'HHmmss"/>-->
<!--Advised：File logging, more configurable-->
<appender name="FILE" class="ch.qos.logback.core.rolling.RollingFileAppender">
<rollingPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
<!-- rollover daily -->
<fileNamePattern>logs/mylog-%d{yyyy-MM-dd}.%i.log</fileNamePattern>
<!-- each file should be at most 100MB, keep 30 days worth of history, but at most 2GB -->
<maxFileSize>100MB</maxFileSize>
<maxHistory>30</maxHistory>
<totalSizeCap>2GB</totalSizeCap>
</rollingPolicy>
<!-- <filter class="ch.qos.logback.classic.filter.LevelFilter"> -->
<filter class="ch.qos.logback.classic.filter.ThresholdFilter">
<level>DEBUG</level>
<!--Options:ACCEPT,NEUTRAL,DENY-->
<!-- <onMatch>ACCEPT</onMatch> -->
<!-- <onMismatch>DENY</onMismatch> -->
</filter>
<immediateFlush>false</immediateFlush>
<append>false</append>
<encoder>
<pattern>%date{ISO8601} %-5level %logger{36} %X{akkaSource} - %msg%n</pattern>
<!--<pattern>%-4relative [%thread] %-5level %logger{35} - %msg%n</pattern>-->
</encoder>
</appender>
<!--For async the logging, drop the info and debug when the block is filled of 80% size by default, maxFlushTime by default-->
<!--<appender name="ASYNC" class="ch.qos.logback.classic.AsyncAppender">-->
<!--<appender-ref ref="FILE" />-->
<!--<nerverBlock>true</nerverBlock>-->
<!--</appender>-->
<!--Deprecated ：Basic file appender configure()-->
<!--<appender name="FILE" class="ch.qos.logback.core.FileAppender">-->
<!--<file>logs/repchain-${bySecond}.log</file>-->
<!--<filter class="ch.qos.logback.classic.filter.ThresholdFilter">-->
<!--<level>WARN</level>-->
<!--</filter>-->
<!--&lt;!&ndash; set immediateFlush to false for much higher logging throughput &ndash;&gt;-->
<!--<immediateFlush>false</immediateFlush>-->
<!--<append>false</append>-->
<!--&lt;!&ndash; encoders are assigned the type-->
<!--ch.qos.logback.classic.encoder.PatternLayoutEncoder by default &ndash;&gt;-->
<!--<encoder>-->
<!--&lt;!&ndash;<pattern>%date{ISO8601} %-5level %logger{36} %X{sourceThread} - %msg%n</pattern>&ndash;&gt;-->
<!--<pattern>%date{ISO8601} %-5level %logger{36} %X{akkaSource} - %msg%n</pattern>-->
<!--&lt;!&ndash;<pattern>%date{yyyy-MM-dd} %X{akkaTimestamp} %-5level[%thread] %logger{1} - %msg%n</pattern>&ndash;&gt;-->
<!--</encoder>-->
<!--</appender>-->
<appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
<target>System.out</target>
<!--level过滤器简化版-->
<filter class="ch.qos.logback.classic.filter.ThresholdFilter">
<level>DEBUG</level>
</filter>
<!--Java 过滤器-->
<!--filter class="ch.qos.logback.core.filter.EvaluatorFilter">
<evaluator>
<expression>return message.contains("Statistic");</expression>
</evaluator>
<OnMismatch>DENY</OnMismatch>
<OnMatch>ACCEPT</OnMatch>
</filter-->
<!--正则表达式过滤器-->
<!--<filter class="ch.qos.logback.core.filter.EvaluatorFilter">-->
<!--<evaluator>-->
<!--<matcher>-->
<!--<Name>pm_1M</Name>-->
<!--&lt;!&ndash; filter out odd numbered statements &ndash;&gt;-->
<!--<regex>block</regex>-->
<!--</matcher>-->
<!--<expression>pm_1M.matches(message)</expression>-->
<!--</evaluator>-->
<!--<OnMismatch>DENY</OnMismatch>-->
<!--<OnMatch>NEUTRAL</OnMatch>-->
<!--</filter>-->
<encoder>
<pattern>%date{ISO8601} %-5level %logger{36} %X{akkaSource} - %msg%n</pattern>
<!--<pattern>%X{akkaTimestamp} %-5level %logger{36} %X{akkaSource} - %msg%n</pattern>-->
<!--<pattern>%X{akkaTimestamp} %-5level[%thread] %logger{0} - %msg%n</pattern>-->
</encoder>
</appender>
<!--<logger name="akka" level="INFO" />-->
<root level="DEBUG">
<appender-ref ref="STDOUT" />
<appender-ref ref="FILE" />
</root>
</configuration>

109
src/main/scala/rep/app/conf/SystemCertList.scala
View File

@ -1,56 +1,55 @@
/*
* Copyright 2018 Blockchain Technology and Application Joint Lab, Linkel Technology Co., Ltd, Beijing, Fintech Research Center of ISCAS.
* 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
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BA SIS,
* 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 rep.app.conf
import java.io._
import rep.crypto.ECDSASign
/**
* @author jiangbuyun
* @version 0.7
* @category 获取信任的证书列表，抽签时从此文件中获取。
* */
object SystemCertList {
private var mySystemCertList:Set[String] = (new scala.collection.mutable.ArrayBuffer[String]()).toSet[String]
private def loadVoteNodeListForCert = {
synchronized{
if(this.mySystemCertList.isEmpty){
val list = SystemProfile.getVoteNodeList
val clist = ECDSASign.getAliasOfTrustkey
var rlist : scala.collection.mutable.ArrayBuffer[String] = new scala.collection.mutable.ArrayBuffer[String]()
var i = 0
for( i <- 1 to clist.size()-1){
val alias = clist.get(i)
if(list.contains(alias)){
rlist += alias
}
}
this.mySystemCertList = rlist.toSet[String]
}
}
}
def getSystemCertList:Set[String] = {
if(this.mySystemCertList.isEmpty){
loadVoteNodeListForCert
}
this.mySystemCertList
}
/*
* Copyright 2018 Blockchain Technology and Application Joint Lab, Linkel Technology Co., Ltd, Beijing, Fintech Research Center of ISCAS.
* 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
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BA SIS,
* 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 rep.app.conf
import java.io._
/**
* @author jiangbuyun
* @version 0.7
* @category 获取信任的证书列表，抽签时从此文件中获取。
* */
object SystemCertList {
private var mySystemCertList:Set[String] = (new scala.collection.mutable.ArrayBuffer[String]()).toSet[String]
private def loadVoteNodeListForCert = {
synchronized{
if(this.mySystemCertList.isEmpty){
val list = SystemProfile.getVoteNodeList
val clist = ECDSASign.getAliasOfTrustkey
var rlist : scala.collection.mutable.ArrayBuffer[String] = new scala.collection.mutable.ArrayBuffer[String]()
var i = 0
for( i <- 1 to clist.size()-1){
val alias = clist.get(i)
if(list.contains(alias)){
rlist += alias
}
}
this.mySystemCertList = rlist.toSet[String]
}
}
}
def getSystemCertList:Set[String] = {
if(this.mySystemCertList.isEmpty){
loadVoteNodeListForCert
}
this.mySystemCertList
}
}

507
src/main/scala/rep/app/system/ClusterSystem.scala
View File

@ -1,254 +1,253 @@
/*
* Copyright 2018 Blockchain Technology and Application Joint Lab, Linkel Technology Co., Ltd, Beijing, Fintech Research Center of ISCAS.
* 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
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BA SIS,
* 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 rep.app.system
import java.io.File
import akka.actor.{ActorRef, ActorSystem, Address, Props}
import akka.cluster.Cluster
import com.typesafe.config.{Config, ConfigFactory}
import rep.app.conf.SystemConf
import rep.app.system.ClusterSystem.InitType
import rep.network.base.ModuleBase
import rep.network.cluster.MemberListener
import rep.network.module.ModuleManager
import rep.network.tools.Statistic.StatisticCollection
import rep.network.tools.register.ActorRegister
import rep.ui.web.EventServer
import rep.utils.GlobalUtils.ActorType
import rep.storage.cfg._
import java.io.File
import scala.collection.mutable
import rep.app.conf.SystemProfile
import com.typesafe.config.ConfigValueFactory
import java.util.List
import java.util.ArrayList
import rep.log.trace.RepLogHelp
import rep.log.trace.LogType
import org.slf4j.LoggerFactory
/**
* System创建伴生对象
* @author shidianyue
* @version 0.7
* @update 2018-05 jiangbuyun
* */
object ClusterSystem {
/**
* 初始化类型
*/
object InitType {
val SINGLE_INIT = 1//单机单节点
val MULTI_INIT = 2//单机多节点
}
private val actorRegisterList = mutable.HashMap[ String, ActorRegister ]()
def register(systemName: String, actorRegister: ActorRegister) = actorRegisterList.put(systemName, actorRegister)
def getActorRegister(sysName: String) = actorRegisterList.get(sysName)
def unregister(systemName: String) = actorRegisterList.remove(systemName)
}
/**
* System创建类
* @author shidianyue
* @version 0.7
* @since 1.0
* @param sysTag 系统system命名
* @param initType 初始化类型
* @param sysStart 是否开启system（不开启仅用于初始化）
* */
class ClusterSystem(sysTag: String, initType: Int, sysStart:Boolean) {
protected def log = LoggerFactory.getLogger(this.getClass)
private val USER_CONFIG_PATH = "conf/system.conf"
private val modulePrefix = "RepCluster"
private val moduleName = modulePrefix + "_" + sysTag
private var webSocket: ActorRef = null
private var memberLis: ActorRef = null
private var moduleManager: ActorRef = null
private var statistics:ActorRef = null
private var enableWebSocket = false
private var enableStatistic = false
private var sysConf: Config = initSystem(sysTag)
private var sysActor:ActorSystem = null
private var clusterAddr:Address = null
/**
* 是否开启Web Socket（API）
*/
def enableWS() = enableWebSocket = true
/**
* 获取用户和系统的联合配置
* @param userConfigFilePath
* @return
*/
def getUserCombinedConf(userConfigFilePath: String): Config = {
val userConfFile = new File(userConfigFilePath)
val innerConf = ConfigFactory.load()
userConfFile.exists() match {
case true =>
val combined_conf = ConfigFactory.parseFile(userConfFile).withFallback(innerConf)
val final_conf = ConfigFactory.load(combined_conf)
final_conf
case false =>
RepLogHelp.logMsg(log,LogType.WARN, moduleName + " ~ " + "Couldn't find the user config file" + " ~ " )
innerConf
}
}
/**
* 获取完整配置信息
* 用户系统初始化
* @param sysName
* @return
*/
def getConfigBySys(sysName: String): Config = {
//TODO 将来找个路径也是可配置的
val myConfig =
ConfigFactory.parseString("akka.remote.netty.ssl.security.key-store = \"jks/mykeystore_" + sysName +
".jks\"")
val regularConfig = getUserCombinedConf(USER_CONFIG_PATH)
val combined =
myConfig.withFallback(regularConfig)
val complete =
ConfigFactory.load(combined)
complete
}
def getConf = sysConf
def hasDiskSpace:Boolean={
var b = true
val sc : StoreConfig = StoreConfig.getStoreConfig()
val ds = sc.getFreeDiskSpace/(1000*1000)
if(SystemProfile.getDiskSpaceAlarm >= ds){
b = false
}
b
}
/**
* 初始化系统参数
* @param sysName
* @return
*/
def initSystem(sysName: String): Config = {
val conf = getConfigBySys(sysName)
RepLogHelp.logMsg(log,LogType.INFO, moduleName + " ~ " + "System configuration successfully" + " ~ " )
enableWebSocket = conf.getInt("system.ws_enable") match {
case 0 => false
case 1 => true
}
enableStatistic = conf.getInt("system.statistic_enable") match {
case 0 => false
case 1 => true
}
conf
}
def getClusterAddr = clusterAddr
/**
* 组网
* @param address
* @return
*/
def joinCluster(address: Address): Boolean = {
initType match {
case InitType.SINGLE_INIT =>
Cluster(sysActor)
case InitType.MULTI_INIT =>
Cluster(sysActor).join(address)
}
true
}
/**
* 初始化
*/
def init = {
initConsensusNodeOfConfig
sysStart match {
case true =>
sysActor = ActorSystem(SystemConf.SYSTEM_NAME, sysConf)
clusterAddr = Cluster(sysActor).selfAddress
case false => //ignore
}
ClusterSystem.register(sysTag, new ActorRegister)
RepLogHelp.logMsg(log,LogType.INFO, "System" + " ~ " + s"System(${sysTag}) init successfully" + " ~ " ,sysTag)
}
private def initConsensusNodeOfConfig={
val nodelist = sysConf.getStringList("system.vote.vote_node_list")
if(nodelist.contains(this.sysTag)){
//val roles = Array("CRFD-Node")
var roles :List[String] = new ArrayList[String]
roles.add("CRFD-Node")
sysConf = sysConf.withValue("akka.cluster.roles", ConfigValueFactory.fromAnyRef(roles))
}
}
/**
* 启动系统
*/
def start = {
if(enableStatistic) statistics = sysActor.actorOf(Props[StatisticCollection],"statistic")
SystemProfile.initConfigSystem(sysActor.settings.config)
moduleManager = sysActor.actorOf(ModuleManager.props("moduleManager", sysTag),"moduleManager")
ModuleBase.registerActorRef(sysTag, ActorType.MODULE_MANAGER, moduleManager)
if(!hasDiskSpace){
Cluster(sysActor).down(clusterAddr)
throw new Exception("not enough disk space")
}
if (enableWebSocket) webSocket = sysActor.actorOf(Props[ EventServer ], "ws")
memberLis = sysActor.actorOf(Props[ MemberListener ], "memberListener")
ModuleBase.registerActorRef(sysTag, ActorType.MEMBER_LISTENER, memberLis)
if (enableWebSocket) ModuleBase.registerActorRef(sysTag, ActorType.API_MODULE, webSocket)
if(enableStatistic) ModuleBase.registerActorRef(sysTag,ActorType.STATISTIC_COLLECTION, statistics)
RepLogHelp.logMsg(log,LogType.INFO, "System" + " ~ " + s"ClusterSystem ${sysTag} start" + " ~ " ,sysTag)
}
/**
* 离网
* @param clusterActor
*/
def leaveCluster(clusterActor:ActorSystem): Unit ={
Cluster(clusterActor).leave(getClusterAddr)
}
def getActorSys:ActorSystem = sysActor
}
/*
* Copyright 2018 Blockchain Technology and Application Joint Lab, Linkel Technology Co., Ltd, Beijing, Fintech Research Center of ISCAS.
* 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
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BA SIS,
* 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 rep.app.system
import java.io.File
import akka.actor.{ActorRef, ActorSystem, Address, Props}
import akka.cluster.Cluster
import com.typesafe.config.{Config, ConfigFactory}
import rep.app.conf.SystemConf
import rep.app.system.ClusterSystem.InitType
import rep.network.base.ModuleBase
import rep.network.cluster.MemberListener
import rep.network.module.ModuleManager
import rep.network.tools.Statistic.StatisticCollection
import rep.network.tools.register.ActorRegister
import rep.ui.web.EventServer
import rep.utils.GlobalUtils.ActorType
import rep.storage.cfg._
import java.io.File
import scala.collection.mutable
import rep.app.conf.SystemProfile
import com.typesafe.config.ConfigValueFactory
import java.util.List
import java.util.ArrayList
import rep.log.trace.LogType
import org.slf4j.LoggerFactory
/**
* System创建伴生对象
* @author shidianyue
* @version 0.7
* @update 2018-05 jiangbuyun
* */
object ClusterSystem {
/**
* 初始化类型
*/
object InitType {
val SINGLE_INIT = 1//单机单节点
val MULTI_INIT = 2//单机多节点
}
private val actorRegisterList = mutable.HashMap[ String, ActorRegister ]()
def register(systemName: String, actorRegister: ActorRegister) = actorRegisterList.put(systemName, actorRegister)
def getActorRegister(sysName: String) = actorRegisterList.get(sysName)
def unregister(systemName: String) = actorRegisterList.remove(systemName)
}
/**
* System创建类
* @author shidianyue
* @version 0.7
* @since 1.0
* @param sysTag 系统system命名
* @param initType 初始化类型
* @param sysStart 是否开启system（不开启仅用于初始化）
* */
class ClusterSystem(sysTag: String, initType: Int, sysStart:Boolean) {
protected def log = LoggerFactory.getLogger(this.getClass)
private val USER_CONFIG_PATH = "conf/system.conf"
private val modulePrefix = "RepCluster"
private val moduleName = modulePrefix + "_" + sysTag
private var webSocket: ActorRef = null
private var memberLis: ActorRef = null
private var moduleManager: ActorRef = null
private var statistics:ActorRef = null
private var enableWebSocket = false
private var enableStatistic = false
private var sysConf: Config = initSystem(sysTag)
private var sysActor:ActorSystem = null
private var clusterAddr:Address = null
/**
* 是否开启Web Socket（API）
*/
def enableWS() = enableWebSocket = true
/**
* 获取用户和系统的联合配置
* @param userConfigFilePath
* @return
*/
def getUserCombinedConf(userConfigFilePath: String): Config = {
val userConfFile = new File(userConfigFilePath)
val innerConf = ConfigFactory.load()
userConfFile.exists() match {
case true =>
val combined_conf = ConfigFactory.parseFile(userConfFile).withFallback(innerConf)
val final_conf = ConfigFactory.load(combined_conf)
final_conf
case false =>
RepLogHelp.logMsg(log,LogType.WARN, moduleName + " ~ " + "Couldn't find the user config file" + " ~ " )
innerConf
}
}
/**
* 获取完整配置信息
* 用户系统初始化
* @param sysName
* @return
*/
def getConfigBySys(sysName: String): Config = {
//TODO 将来找个路径也是可配置的
val myConfig =
ConfigFactory.parseString("akka.remote.netty.ssl.security.key-store = \"jks/mykeystore_" + sysName +
".jks\"")
val regularConfig = getUserCombinedConf(USER_CONFIG_PATH)
val combined =
myConfig.withFallback(regularConfig)
val complete =
ConfigFactory.load(combined)
complete
}
def getConf = sysConf
def hasDiskSpace:Boolean={
var b = true
val sc : StoreConfig = StoreConfig.getStoreConfig()
val ds = sc.getFreeDiskSpace/(1000*1000)
if(SystemProfile.getDiskSpaceAlarm >= ds){
b = false
}
b
}
/**
* 初始化系统参数
* @param sysName
* @return
*/
def initSystem(sysName: String): Config = {
val conf = getConfigBySys(sysName)
RepLogHelp.logMsg(log,LogType.INFO, moduleName + " ~ " + "System configuration successfully" + " ~ " )
enableWebSocket = conf.getInt("system.ws_enable") match {
case 0 => false
case 1 => true
}
enableStatistic = conf.getInt("system.statistic_enable") match {
case 0 => false
case 1 => true
}
conf
}
def getClusterAddr = clusterAddr
/**
* 组网
* @param address
* @return
*/
def joinCluster(address: Address): Boolean = {
initType match {
case InitType.SINGLE_INIT =>
Cluster(sysActor)
case InitType.MULTI_INIT =>
Cluster(sysActor).join(address)
}
true
}
/**
* 初始化
*/
def init = {
initConsensusNodeOfConfig
sysStart match {
case true =>
sysActor = ActorSystem(SystemConf.SYSTEM_NAME, sysConf)
clusterAddr = Cluster(sysActor).selfAddress
case false => //ignore
}
ClusterSystem.register(sysTag, new ActorRegister)
RepLogHelp.logMsg(log,LogType.INFO, "System" + " ~ " + s"System(${sysTag}) init successfully" + " ~ " ,sysTag)
}
private def initConsensusNodeOfConfig={
val nodelist = sysConf.getStringList("system.vote.vote_node_list")
if(nodelist.contains(this.sysTag)){
//val roles = Array("CRFD-Node")
var roles :List[String] = new ArrayList[String]
roles.add("CRFD-Node")
sysConf = sysConf.withValue("akka.cluster.roles", ConfigValueFactory.fromAnyRef(roles))
}
}
/**
* 启动系统
*/
def start = {
if(enableStatistic) statistics = sysActor.actorOf(Props[StatisticCollection],"statistic")
SystemProfile.initConfigSystem(sysActor.settings.config)
moduleManager = sysActor.actorOf(ModuleManager.props("moduleManager", sysTag),"moduleManager")
ModuleBase.registerActorRef(sysTag, ActorType.MODULE_MANAGER, moduleManager)
if(!hasDiskSpace){
Cluster(sysActor).down(clusterAddr)
throw new Exception("not enough disk space")
}
if (enableWebSocket) webSocket = sysActor.actorOf(Props[ EventServer ], "ws")
memberLis = sysActor.actorOf(Props[ MemberListener ], "memberListener")
ModuleBase.registerActorRef(sysTag, ActorType.MEMBER_LISTENER, memberLis)
if (enableWebSocket) ModuleBase.registerActorRef(sysTag, ActorType.API_MODULE, webSocket)
if(enableStatistic) ModuleBase.registerActorRef(sysTag,ActorType.STATISTIC_COLLECTION, statistics)
RepLogHelp.logMsg(log,LogType.INFO, "System" + " ~ " + s"ClusterSystem ${sysTag} start" + " ~ " ,sysTag)
}
/**
* 离网
* @param clusterActor
*/
def leaveCluster(clusterActor:ActorSystem): Unit ={
Cluster(clusterActor).leave(getClusterAddr)
}
def getActorSys:ActorSystem = sysActor
}

15
src/main/scala/rep/crypto/SignFunc.scala → src/main/scala/rep/crypto/cert/ISigner.scala
View File

@ -14,15 +14,20 @@
*
*/
package rep.crypto
package rep.crypto.cert
import java.security.{PrivateKey,PublicKey}
import java.security.cert.{ Certificate, CertificateFactory }
import java.security._
/**
* @author c4w
* 系统签名相关提交给外部的接口，第三方使用不需要直接调用该类
* @author jiangbuyun
* @version 1.0
*/
trait SignFunc {
trait ISigner {
def sign(privateKey: PrivateKey, message: Array[Byte]): Array[Byte]
def verify(signature: Array[Byte], message: Array[Byte], publicKey: PublicKey): Boolean
def getCertWithCheck(certAddr:String,certKey:String,sysTag:String):Option[java.security.cert.Certificate]
def CertificateIsValid(date:java.util.Date, cert:Certificate):Boolean
}

86
src/main/scala/rep/crypto/cert/ImpECDSASigner.scala Normal file
View File

@ -0,0 +1,86 @@
/*
* Copyright 2018 Blockchain Technology and Application Joint Lab, Linkel Technology Co., Ltd, Beijing, Fintech Research Center of ISCAS.
* 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
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BA SIS,
* 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 rep.crypto.cert
import java.security._
import java.io._
import java.security.cert.{ Certificate, CertificateFactory }
import rep.app.conf.SystemProfile
import com.google.protobuf.ByteString
import fastparse.utils.Base64
import rep.utils.SerializeUtils
import rep.storage._
import scala.collection.mutable
import com.fasterxml.jackson.core.Base64Variants
import java.security.cert.X509Certificate
import javax.xml.bind.DatatypeConverter
import java.util.ArrayList
import java.util.List
import sun.security.ec.ECPublicKeyImpl
/**
* 实现系统签名和验签功能，第三方使用不需要直接调用该类
* @author jiangbuyun
* @version 1.0
*/
class ImpECDSASigner extends ISigner {
private val alg = "SHA1withECDSA"
override def sign(privateKey: PrivateKey, message: Array[Byte]): Array[Byte] = {
if(privateKey == null) throw new RuntimeException("签名时私钥为空！")
val s1 = Signature.getInstance(alg);
s1.initSign(privateKey)
s1.update(message)
s1.sign()
}
override def verify(signature: Array[Byte], message: Array[Byte], publicKey: PublicKey): Boolean = {
if(publicKey == null) throw new RuntimeException("验证签名时公钥为空！")
val s2 = Signature.getInstance(alg);
s2.initVerify(publicKey)
s2.update(message)
s2.verify(signature)
}
override def CertificateIsValid(date:java.util.Date, cert:Certificate):Boolean={
var isValid :Boolean = false
var start = System.currentTimeMillis()
try {
if(cert == null){
isValid = false
}else{
if(SystemProfile.getCheckCertValidate == 0){
isValid = true
}else if(SystemProfile.getCheckCertValidate == 1){
if(cert.isInstanceOf[X509Certificate]){
var x509cert :X509Certificate = cert.asInstanceOf[X509Certificate]
x509cert.checkValidity(date)
isValid = true
}
}else{
isValid = true
}
}
} catch{
case e : Exception => isValid = false
}
var end = System.currentTimeMillis()
//println("check cert validate,spent time="+(end-start))
isValid;
}
}

145
src/main/scala/rep/crypto/cert/SignTool.scala Normal file
View File

@ -0,0 +1,145 @@
/*
* Copyright 2018 Blockchain Technology and Application Joint Lab, Linkel Technology Co., Ltd, Beijing, Fintech Research Center of ISCAS.
* 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
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BA SIS,
* 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 rep.crypto.cert
import java.security.{ PrivateKey, PublicKey, KeyStore }
import java.security.cert.{ Certificate, CertificateFactory }
import rep.protos.peer.CertId
import scala.collection.mutable
import java.io._
import java.util.{ ArrayList, List }
/**
* 负责签名和验签的工具了，所有相关的功能都调用该类
* @author jiangbuyun
* @version 1.0
*/
object SignTool {
private var signer: ISigner = null
private var SignType: String = "ECDSA"
private var key_password = mutable.HashMap[String, String]()
private var keyStore = mutable.HashMap[String, KeyStore]()
private var PublickeyCerts = mutable.HashMap[String, Certificate]()
synchronized {
if (this.signer == null) {
signer = new ImpECDSASigner()
}
}
private def getPrivateKey(alias: String): PrivateKey = {
val sk = keyStore(alias).getKey(alias, key_password(alias).toCharArray)
sk.asInstanceOf[PrivateKey]
}
private def getPrivateKey(certinfo: CertId): PrivateKey = {
//todo
null
}
//根据CertId实现签名
def sign(certinfo: CertId, message: Array[Byte]): Array[Byte] = {
val key = certinfo.creditCode + "_" + certinfo.certName
var pk: PrivateKey = null
if (this.keyStore.contains(key)) {
pk = getPrivateKey(key)
} else {
pk = getPrivateKey(certinfo)
}
this.signer.sign(pk, message)
}
//根据私钥实现签名
def sign(privateKey: PrivateKey, message: Array[Byte]): Array[Byte] = {
this.signer.sign(privateKey, message)
}
private def getPublicKeyByName(alias: String): PublicKey = {
var tmpcert = PublickeyCerts.get(alias)
if (tmpcert == null && tmpcert != None) {
throw new RuntimeException("证书不存在")
}
if (this.signer.CertificateIsValid(new java.util.Date(), tmpcert.get)) {
tmpcert.get.getPublicKey
} else {
null
}
}
private def getPublicKeyByName(certinfo: CertId): PublicKey = {
//todo
null
}
//根据CertId实现验签
def verify(signature: Array[Byte], message: Array[Byte], certinfo: CertId): Boolean = {
val key = certinfo.creditCode + "_" + certinfo.certName
var pubkey: PublicKey = null
pubkey = getPublicKeyByName(key)
if (pubkey == null) {
pubkey = getPublicKeyByName(certinfo)
}
this.signer.verify(signature, message, pubkey)
}
//根据公钥实现签名
def verify(signature: Array[Byte], message: Array[Byte], publicKey: PublicKey): Boolean = {
this.signer.verify(signature, message, publicKey)
}
//节点启动时需要调用该函数初始化节点私钥
def InitNodePrivateKey(alias: String, password: String, path: String) = {
synchronized {
key_password(alias + "_1") = password
val fis = new FileInputStream(new File(path))
val pwd = password.toCharArray()
if (keyStore.contains(alias)) {
keyStore(alias).load(fis, pwd)
} else {
val pkeys = KeyStore.getInstance(KeyStore.getDefaultType)
pkeys.load(fis, pwd)
keyStore(alias + "_1") = pkeys
}
}
}
//节点启动时需要调用该函数初始化节点公钥
def InitNodePublicKey(password: String, path: String) = {
synchronized {
val fis = new FileInputStream(new File(path))
val pwd = password.toCharArray()
var trustKeyStore = KeyStore.getInstance(KeyStore.getDefaultType)
trustKeyStore.load(fis, pwd)
val enums = trustKeyStore.aliases()
while (enums.hasMoreElements) {
val alias = enums.nextElement()
val cert = trustKeyStore.getCertificate(alias)
PublickeyCerts.put(alias + "_1", cert)
}
}
}
//提供给共识获取证书列表
def getAliasOfTrustkey: List[String] = {
var list: List[String] = new ArrayList[String]
val enums = PublickeyCerts.iterator
while (enums.hasNext) {
val alias = enums.next()
list.add(alias._1)
}
list
}
}

51
src/main/scala/rep/log/trace/LogOption.scala Normal file
View File

@ -0,0 +1,51 @@
/*
* Copyright 2018 Blockchain Technology and Application Joint Lab, Linkel Technology Co., Ltd, Beijing, Fintech Research Center of ISCAS.
* 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
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BA SIS,
* 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 rep.log.trace
import java.util.concurrent.ConcurrentHashMap
import scala.collection.JavaConverters._
import rep.log.trace.ModuleType.ModuleType;
object LogOption {
private implicit var options = new ConcurrentHashMap[String, Boolean] asScala
def isPrintLog(nodeName:String,mt:ModuleType):Boolean={
var b = true
if(options.contains(nodeName)){
if(options(nodeName)){
val k = nodeName+"_"+mt.toString()
if(options.contains(k)){
b = options(k)
}
}else{
b = false
}
}
b
}
def openNodeLog(nodeName:String)={
options.put(nodeName, true)
}
def closeNodeLog(nodeName:String)={
options.put(nodeName, false)
}
def setModuleLogOption(nodeName:String,option:String,isOpen:Boolean)={
options.put(nodeName+"_"+option, isOpen)
}
}

146
src/main/scala/rep/log/trace/LogTraceOption.java
View File

@ -1,146 +0,0 @@
/*
* Copyright 2018 Blockchain Technology and Application Joint Lab, Linkel Technology Co., Ltd, Beijing, Fintech Research Center of ISCAS.
* 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
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BA SIS,
* 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 rep.log.trace;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.StringTokenizer;
public class LogTraceOption {
private static LogTraceOption traceOptionObj = null;
private ConcurrentHashMap<String,Boolean> lhm = null;
private ConcurrentHashMap<String,Boolean> nodeLog = null;
private AtomicBoolean isOpen = null;
private LogTraceOption() {
this.lhm = new ConcurrentHashMap<String,Boolean>();
this.nodeLog = new ConcurrentHashMap<String,Boolean>();
this.isOpen = new AtomicBoolean(true);
}
public static synchronized LogTraceOption getLogTraceOption() {
if(traceOptionObj == null) {
traceOptionObj = new LogTraceOption();
}
return traceOptionObj;
}
public void addLogOption(String packageName,Boolean status) {
lhm.put(packageName, status);
}
public void addNodeLogOption(String nodeName,Boolean status) {
nodeLog.put(nodeName, status);
}
public boolean isOpenOutputLog(String nodeName,String packageName) {
boolean rb = true;
if(this.isOpen.get()) {
if(nodeName != null) {
if(isNodeClose(nodeName)) {
rb = isClose(packageName);
}else {
rb = false;
}
}else {
rb = isClose(packageName);
}
}else {
rb = false;
}
return rb;
}
private boolean isNodeClose(String nodeNme) {
boolean isclose = true;
if(this.nodeLog.isEmpty()) {
return isclose;
}else {
if(!this.nodeLog.containsKey(nodeNme)) {
return isclose;
}else {
Boolean b = this.nodeLog.get(nodeNme);
isclose = b.booleanValue();
return isclose;
}
}
}
private boolean isClose(String packageName) {
boolean isclose = true;
if(this.lhm.isEmpty()) {
return isclose;
}else {
String pn = packageName;
while(!pn.equalsIgnoreCase("")) {
if(!this.lhm.containsKey(pn)) {
pn = getParent(pn);
}else {
Boolean b = this.lhm.get(pn);
isclose = b.booleanValue();
break;
}
}
return isclose;
}
}
/*private boolean getValueForName(String packageName) {
if(!this.lhm.containsKey(packageName)) {
return false;
}else {
Boolean b = this.lhm.get(packageName);
return b.booleanValue();
}
}*/
private String getParent(String name) {
String rs = "";
StringTokenizer st = new StringTokenizer(name,".",false);
if(st.countTokens() == 1) {
return rs;
}else {
int len = st.countTokens() - 1;
for(int i = 0; i < len; i++) {
if(i == 0) {
rs = rs + st.nextToken();
}else {
rs = rs + "."+st.nextToken();
}
}
}
return rs;
}
public void CloseAll() {
this.isOpen.set(false);
}
public void OpenAll() {
this.isOpen.set(true);
}
}

24
src/main/scala/rep/storage/Storage.scala → src/main/scala/rep/log/trace/LogType.scala
View File

@ -13,20 +13,18 @@
* limitations under the License.
*
*/
package rep.storage
package rep.log.trace
/**
* @author c4w
* 定义日志的类型
* @author jiangbuyun
* @version 1.0
*/
trait Storage[Key, Value] {
def set(key: Key, value: Value): Unit
def get(key: Key): Option[Value]
def containsKey(key: Key): Boolean = get(key).isDefined
def commit(): Unit
def merkle():Array[Byte]
}
object LogType extends Enumeration{
type LogType = Value
val INFO = Value(1)
val WARN = Value(2)
val DEBUG = Value(3)
val ERROR = Value(4)
}

22
src/main/scala/rep/log/trace/LogType.java → src/main/scala/rep/log/trace/ModuleType.scala
View File

@ -13,9 +13,23 @@
* limitations under the License.
*
*/
package rep.log.trace
package rep.log.trace;
/**
* 定义输出日志的模块
* @author jiangbuyun
* @version 1.0
*/
public enum LogType {
INFO,DEBUG,WARN,ERROR
}
object ModuleType extends Enumeration{
type ModuleType = Value
val blocker = Value("blocker")
val endorser = Value("endorser")
val timeTracer = Value("timeTracer")
val storager = Value("storager")
val preloadTrans = Value("preloadTrans")
val contract = Value("contract")
val others = Value("others")
def checkExists(t:String) = this.values.exists(_.toString==t)
}

58
src/main/scala/rep/log/trace/RepLogHelp.scala
View File

@ -1,58 +0,0 @@
/*
* Copyright 2018 Blockchain Technology and Application Joint Lab, Linkel Technology Co., Ltd, Beijing, Fintech Research Center of ISCAS.
* 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
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BA SIS,
* 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 rep.log.trace
import org.slf4j.LoggerFactory
import org.slf4j.Logger;
object RepLogHelp {
case object LOG_TYPE{
val INFO = 1
val DEBUG =2
val WARN = 3
val ERROR = 4
}
def logMsg(log: Logger,lOG_TYPE: LogType, msg:String, nodeName:String) = {
lOG_TYPE match {
case LogType.INFO =>
log.info(msg,Array(nodeName))
case LogType.DEBUG =>
log.debug(msg,Array(nodeName))
case LogType.WARN =>
log.warn(msg,Array(nodeName))
case LogType.ERROR =>
log.error(msg,Array(nodeName))
}
}
def logMsg(log: Logger,lOG_TYPE: LogType, msg:String) = {
lOG_TYPE match {
case LogType.INFO =>
log.info(msg)
case LogType.DEBUG =>
log.debug(msg)
case LogType.WARN =>
log.warn(msg)
case LogType.ERROR =>
log.error(msg)
}
}
}

72
src/main/scala/rep/log/trace/RepLogger.scala Normal file
View File

@ -0,0 +1,72 @@
/*
* Copyright 2018 Blockchain Technology and Application Joint Lab, Linkel Technology Co., Ltd, Beijing, Fintech Research Center of ISCAS.
* 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
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BA SIS,
* 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 rep.log.trace
import org.slf4j.LoggerFactory
import org.slf4j.Logger;
import rep.log.trace.ModuleType.ModuleType
import rep.log.trace.LogType.LogType
/**
* RepChain统一日志输出工具，外部输出日志统一调用对象
* @author jiangbuyun
* @version 1.0
*/
object RepLogger {
protected def log = LoggerFactory.getLogger(this.getClass)
def logWarn(nodeName:String,mtype:ModuleType,message:String) = {
if(LogOption.isPrintLog(nodeName, mtype)){
logMsg(LogType.WARN,getMessage(nodeName,mtype,message) )
}
}
def logDebug(nodeName:String,mtype:ModuleType,message:String) = {
if(LogOption.isPrintLog(nodeName, mtype)){
logMsg(LogType.DEBUG,getMessage(nodeName,mtype,message) )
}
}
def logInfo(nodeName:String,mtype:ModuleType,message:String) = {
if(LogOption.isPrintLog(nodeName, mtype)){
logMsg(LogType.INFO,getMessage(nodeName,mtype,message) )
}
}
def logError(nodeName:String,mtype:ModuleType,message:String) = {
if(LogOption.isPrintLog(nodeName, mtype)){
logMsg(LogType.ERROR,getMessage(nodeName,mtype,message))
}
}
private def getMessage(nodeName:String,mtype:ModuleType,message:String):String={
nodeName + " ~ " +mtype.toString() + " ~ " + message
}
private def logMsg(logtype: LogType.LogType, msg:String) = {
logtype match {
case LogType.INFO =>
log.info(msg)
case LogType.DEBUG =>
log.debug(msg)
case LogType.WARN =>
log.warn(msg)
case LogType.ERROR =>
log.error(msg)
}
}
}

64
src/main/scala/rep/log/trace/RepTimeTrace.java
View File

@ -1,64 +0,0 @@
/*
* Copyright 2018 Blockchain Technology and Application Joint Lab, Linkel Technology Co., Ltd, Beijing, Fintech Research Center of ISCAS.
* 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
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BA SIS,
* 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 rep.log.trace;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
public class RepTimeTrace {
private static RepTimeTrace timetrace = null;
private ConcurrentHashMap<String,TimePair> nodeTimeTracer = null;
private AtomicBoolean isOpenTimeTrace = null;
private RepTimeTrace() {
this.nodeTimeTracer = new ConcurrentHashMap<String,TimePair>();
this.isOpenTimeTrace = new AtomicBoolean(false);
}
public static synchronized RepTimeTrace getRepTimeTrace() {
if(timetrace == null) {
timetrace = new RepTimeTrace();
}
return timetrace;
}
public void addTimeTrace(String nodeName,String timetag,long time,boolean isstart) {
if(this.isOpenTimeTrace.get()) {
TimePair rtd = null;
if(this.nodeTimeTracer.containsKey(nodeName+"_"+timetag)) {
rtd = this.nodeTimeTracer.get(nodeName+"_"+timetag);
}else {
rtd = new TimePair(nodeName+"_"+timetag,nodeName);
this.nodeTimeTracer.put(nodeName+"_"+timetag,rtd);
}
if(isstart) {
rtd.start(time);
}else {
rtd.finish(time);
}
}
}
public void openTimeTrace() {
this.isOpenTimeTrace.set(true);
}
public void closeTimeTrace() {
this.isOpenTimeTrace.set(false);
}
}

43
src/main/scala/rep/log/LogFilterExp.java → src/main/scala/rep/log/trace/RepTimeTracer.scala
View File

@ -13,28 +13,31 @@
* limitations under the License.
*
*/
package rep.log.trace
package rep.log;
import java.util.concurrent.ConcurrentHashMap
import scala.collection.JavaConverters._
/**
* Created by User on 2017/7/23.
*/
import ch.qos.logback.classic.spi.ILoggingEvent;
import ch.qos.logback.core.filter.Filter;
import ch.qos.logback.core.spi.FilterReply;
/**
* log过滤类实现
* @author shidianyue
* RepChain系统运行时间跟踪工具，需要跟踪运行时间的程序统一调用该对象
* @author jiangbuyun
* @version 1.0
* */
public class LogFilterExp extends Filter<ILoggingEvent> {
*/
@Override
public FilterReply decide(ILoggingEvent event) {
if (event.getMessage().contains("Opt Time")) {
return FilterReply.ACCEPT;
} else {
return FilterReply.DENY;
}
}
}
object RepTimeTracer {
private implicit var times = new ConcurrentHashMap[String, Long] asScala
def setStartTime(nodeName:String,flag:String,t:Long)={
this.times.put(nodeName+"-"+flag, t);
}
def setEndTime(nodeName:String,flag:String,t:Long)={
val key = nodeName+"-"+flag;
if(this.times.contains(key)) {
val tl = t - this.times(key);
RepLogger.logInfo(nodeName, ModuleType.timeTracer, key+"="+tl)
}
}
}

56
src/main/scala/rep/log/trace/TimePair.java
View File

@ -1,56 +0,0 @@
/*
* Copyright 2018 Blockchain Technology and Application Joint Lab, Linkel Technology Co., Ltd, Beijing, Fintech Research Center of ISCAS.
* 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
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BA SIS,
* 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 rep.log.trace;
import org.slf4j.LoggerFactory;
import org.slf4j.Logger;
public class TimePair {
private static Logger log = LoggerFactory.getLogger(TimePair.class);
private String timeTag = "";
private String nodeName = "";
private long start = 0l;
private long end = 0;
public TimePair(String timeTag,String nodeName) {
this.timeTag = timeTag;
this.nodeName = nodeName;
}
public void start(long start) {
this.start = start;
this.end = 0l;
}
public synchronized void finish(long end) {
this.end = end;
String str = this.toString();
//if(nodeName.equalsIgnoreCase("1")) {
log.info(str);
System.out.println(str);
//}
this.start = 0l;
this.end = 0l;
}
public String toString() {
if(this.start > 0 && this.end > 0)
return "~"+"timeTag="+this.timeTag+"~"+"start="+(this.start)+"~"+"end="+this.end+"~"+"timespan="+(this.end-this.start);
else
return "~"+"timeTag="+this.timeTag+"~"+" not enough time item";
}
}

69
src/main/scala/rep/log/trace/repTurboLogFilter.java
View File

@ -1,69 +0,0 @@
/*
* Copyright 2018 Blockchain Technology and Application Joint Lab, Linkel Technology Co., Ltd, Beijing, Fintech Research Center of ISCAS.
* 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
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BA SIS,
* 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 rep.log.trace;
import org.slf4j.Marker;
import ch.qos.logback.classic.Level;
import ch.qos.logback.classic.Logger;
import ch.qos.logback.classic.turbo.TurboFilter;
import ch.qos.logback.core.spi.FilterReply;
import rep.log.trace.LogTraceOption;
public class repTurboLogFilter extends TurboFilter {
private LogTraceOption logOption = LogTraceOption.getLogTraceOption();
@Override
public FilterReply decide(Marker marker, Logger logger, Level level,
String format, Object[] params, Throwable t) {
if (!isStarted()) {
return FilterReply.NEUTRAL;
}
if(params != null && params.length > 0) {
Object sysname = params[0];
if(sysname != null) {
String nameobj = null;
if(sysname instanceof String[]) {
nameobj = ((String[])sysname)[0];
}
if(logOption.isOpenOutputLog(nameobj,logger.getName())) {
return FilterReply.ACCEPT;
}else {
return FilterReply.DENY;
}
}else {
if(logOption.isOpenOutputLog(null,logger.getName())) {
return FilterReply.ACCEPT;
}else {
return FilterReply.DENY;
}
}
}else {
if(logOption.isOpenOutputLog(null,logger.getName())) {
return FilterReply.ACCEPT;
}else {
return FilterReply.DENY;
}
}
}
@Override
public void start() {
super.start();
}
}

51
src/main/scala/rep/log/trace/test.scala Normal file
View File

@ -0,0 +1,51 @@
/*
* Copyright 2018 Blockchain Technology and Application Joint Lab, Linkel Technology Co., Ltd, Beijing, Fintech Research Center of ISCAS.
* 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
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BA SIS,
* 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 rep.log.trace
import rep.log.trace.ModuleType.ModuleType
/**
* 日志模块的测试代码
* @author jiangbuyun
* @version 1.0
*/
object test extends Object{
def main(args: Array[String]): Unit = {
var i = 0
while(i < 100){
RepLogger.logInfo("1", ModuleType.endorser, s"test{$i}")
if(i == 10){
LogOption.closeNodeLog("1")
}
if(i == 20){
LogOption.openNodeLog("1")
LogOption.setModuleLogOption("1", "endorser", false)
}
if(i == 30){
LogOption.setModuleLogOption("1", "endorser", true)
}
if(i == 35){
RepTimeTracer.setStartTime("1", "test", System.currentTimeMillis())
}
if(i == 37){
RepTimeTracer.setEndTime("1", "test", System.currentTimeMillis())
}
i = i+1
}
}
}

52
src/main/scala/rep/storage/FakeStorage.scala
View File

@ -1,52 +0,0 @@
/*
* Copyright 2018 Blockchain Technology and Application Joint Lab, Linkel Technology Co., Ltd, Beijing, Fintech Research Center of ISCAS.
* 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
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BA SIS,
* 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 rep.storage
import rep.protos._
import rep.log.trace.RepLogHelp
import rep.log.trace.LogType
import org.slf4j.LoggerFactory
/**
* @author c4w
*/
class FakeStorage() extends Storage[String,Array[Byte]] {
import FakeStorage._
protected def log = LoggerFactory.getLogger(this.getClass)
private val m = scala.collection.mutable.Map[Key,Value]()
override def set(key:Key, value:Value): Unit = {
val me = this
log.info(s"set state:$key $value $me")
m.put(key,value)
}
override def get(key:Key): Option[Value] = {
m.get(key)
}
override def commit(): Unit={
}
override def merkle():Array[Byte]={
null
}
}
object FakeStorage {
type Key = String
type Value = Array[Byte]
}

1217
src/main/scala/rep/storage/IDataAccess.scala
View File

File diff suppressed because it is too large Load Diff

1564
src/main/scala/rep/storage/ImpDataAccess.scala
View File

File diff suppressed because it is too large Load Diff

966
src/main/scala/rep/storage/ImpDataPreload.scala
View File

@ -1,478 +1,490 @@
/*
* Copyright 2018 Blockchain Technology and Application Joint Lab, Linkel Technology Co., Ltd, Beijing, Fintech Research Center of ISCAS.
* 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
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BA SIS,
* 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 rep.storage
import rep.utils._
import scala.collection.mutable
import rep.storage.leveldb._
import rep.storage.merkle._
import scala.collection.mutable.ArrayBuffer
import rep.protos.peer._;
import scala.util.control.Breaks
/**内存数据库的访问类，属于多实例。
* @constructor 根据SystemName和InstanceName建立实例
* @param SystemName 系统名
* @param InstanceName 实例名
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* @category 内存数据库的访问类，属于多实例。
*/
class ImpDataPreload (SystemName:String,InstanceName:String) extends AbstractLevelDB {
private var update :java.util.concurrent.ConcurrentHashMap[String,Array[Byte]] = new java.util.concurrent.ConcurrentHashMap[String,Array[Byte]]
private var dbop = ImpDataAccess.GetDataAccess(SystemName)
//private var merkleop : RepBucket = new RepBucket(this)
this.ReloadMerkle
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* @category 获取当前系统的名称
* @param 无
* @return 返回当前系统的名称 String
* */
override def getSystemName:String={
SystemName
}
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* @category 获取当前实例的名称
* @param 无
* @return 返回当前实例的名称 String
* */
override def getInstanceName:String={
InstanceName
}
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* @category 获取指定的键值
* @param key String 指定的键
* @return 返回对应键的值 Array[Byte]
* */
override def Get(key : String):Array[Byte]={
var rb : Array[Byte] = null
try{
if(this.update.containsKey(key)){
rb = this.update.get(key)
}else{
rb = this.dbop.Get(key)
}
setUseTime
}catch{
case e:Exception =>{
rb = null
log.error("ImpDataPreload_" + SystemName + "_" + s"ImpDataPreload Get failed, error info= "+e.getMessage)
throw e
}
}
rb
}
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* @category 存储指定的键和值到数据库
* @param key String 指定的键，bb Array[Byte] 要存储的值
* @return 返回成功或者失败 Boolean
* */
override def Put (key : String,bb : Array[Byte],isWorldState : Boolean):Boolean={
var b : Boolean = true
try{
if(key == null){
log.info("ImpDataPreload_" + SystemName + "_" + s"ImpDataPreload Put failed, error info= key is null")
}
if(bb == null){
log.info("ImpDataPreload_" + SystemName + "_" + s"ImpDataPreload Put failed, error info= value is null")
}
if(key != null && bb != null){
if(this.update.get(key) != null){
val obbstr = toString(this.update.get(key))
val bbstr = toString(bb)
if(!obbstr.equals(bbstr)){
this.update.put(key, bb)
//this.merkleop.Put(key, bb)
this.PutWorldStateToMerkle(key,bb)
}
}else{
if(this.dbop.Get(key) != null){
val obbstr = toString(this.dbop.Get(key))
val bbstr = toString(bb)
if(!obbstr.equals(bbstr)){
this.update.put(key, bb)
//this.merkleop.Put(key, bb)
this.PutWorldStateToMerkle(key,bb)
}
}else{
this.update.put(key, bb)
//this.merkleop.Put(key, bb)
this.PutWorldStateToMerkle(key,bb)
}
}
}
setUseTime
}catch{
case e:Exception =>{
b = false
log.error("ImpDataPreload_" + SystemName + "_" + s"ImpDataPreload Put failed, error info= "+e.getMessage)
throw e
}
}
b
}
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* @category 删除指定的键值
* @param key String 指定的键
* @return 返回成功或者失败 Boolean
* 该类暂时没有实现，因为RepChain不能够删除已有的WorldState
* */
override def Delete (key : String) : Boolean={
var b : Boolean = true;
//todo 目前没有实现，原则上不允许删除任何状态
setUseTime
b;
}
private var useTime:Long = System.currentTimeMillis();
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* @category 内部使用，更新实例的访问时间，每次访问的时候都会调用该方法
* @param key String 指定的键
* @return 无
* */
def setUseTime{
this.useTime = System.currentTimeMillis();
}
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* @category 内部使用，获取该实例最后一次使用时间
* @param 无
* @return 长整型，最后一次的使用时间
* */
def getUseTime:Long={
this.useTime
}
//////////////////////////////Endorsement 背书验证/////////////////////////////////////////////////////////////////////////////
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* @category 根据交易id获取chaincode id
* @param block Block 待写入的区块,txid String 待查询的交易id
* @return 如果成功返回chaincode id，否则返回空字符串
* */
private def getTxidFormBlock(block:Block,txid:String):String={
var rel = ""
if(block != null){
var trans = block.transactions
if(trans.length > 0){
val loopbreak = new Breaks
loopbreak.breakable(
trans.foreach(f=>{
if(f.txid.equals(txid)){
val chainspec = f.payload.get
rel = chainspec.chaincodeID.get.name
loopbreak.break
}
})
)
}
}
rel
}
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* @category 交易背书时调用，用于验证读写指令集
* @param block Block 待验证的区块
* @return 如果验证成功返回true，否则false
* */
def VerifyForEndorsement(block:Block):Boolean={
var b : Boolean = false
if(block == null) return b
if(block.stateHash == null) return b
try{
var list : scala.collection.mutable.LinkedHashMap[String,Seq[OperLog]] = new scala.collection.mutable.LinkedHashMap[String,Seq[OperLog]]()
val txresults = block.getNonHashData.transactionResults
txresults.foreach(f=>{
list += f.txid -> f.ol
}
)
if(list.size > 0){
list.foreach(f=>{
val txid = f._1
val cid = getTxidFormBlock(block,txid)
val jobj = f._2
if(jobj != null && jobj.length > 0){
jobj.foreach(f=>{
var tmpkeystr = IdxPrefix.WorldStateKeyPreFix+cid+"_"+f.key
this.Put(IdxPrefix.WorldStateKeyPreFix+cid+"_"+f.key, f.newValue.toByteArray())
})
}
})
}
val shash4block = block.stateHash.toStringUtf8
val shash4local = this.GetComputeMerkle4String
if(shash4block.equals(shash4local)){
b = true
}else{
b = false
this.log.error("system_name="+this.SystemName+"\t verify World State is failed, hash4block="+shash4block+"\t hash4local="+shash4local)
}
}catch{
case e:Exception =>{
throw e;
}
}finally{
//todo
}
b
}
/////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
////////////////////////////////worldstate merkle 继承实现///////////////////////////////////////////////////////////////////////
/*override def GetComputeMerkle:Array[Byte]={
setUseTime
val b = this.merkleop.getMerkleHash
b
}
override def GetComputeMerkle4String:String={
setUseTime
val c = this.merkleop.getMerkelHash4String
c
}*/
///////////////////////////////////////////////////////////////////////////////////////////////////////
}
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* @category 某个系统的多实例管理类，管理某个系统所有的多实例，包含自动释放超时的实例。
* */
private class MultiDBMgr (val SystemName:String) {
import org.slf4j.LoggerFactory
private val checktime = 60*1000//如果某个实例超过60s没有使用，就自动清理
private var DBOps = new scala.collection.mutable.HashMap[String,ImpDataPreload]()
protected def log = LoggerFactory.getLogger(this.getClass)
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* @category 根据实例名称获取预执行的数据访问类
* @param InstanceName String 实例名称
* @return 如果成功返回ImpDataPreload实例，否则为null
* */
def GetImpDataPreload( InstanceName:String) : ImpDataPreload = {
var DBOp : ImpDataPreload = null
synchronized{
try{
clear
if(DBOps.contains(InstanceName)){
DBOp = DBOps(InstanceName)
}else{
DBOp = new ImpDataPreload(SystemName,InstanceName)
DBOps.put(InstanceName, DBOp)
}
if(DBOp != null){
DBOp.setUseTime
}
}catch{
case e:Exception =>{
log.error("MultiDBMgr_" + SystemName + "_" + s"ImpDataPreload Create failed, error info= "+e.getMessage)
throw e
}
}
DBOp
}
}
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* @category 清理系统中空闲超时的实例
* @param 无
* @return 无
* */
def clear{
var exists : ArrayBuffer[String] = new ArrayBuffer[String]()
val iterator = DBOps.keysIterator
while (iterator.hasNext) {
val key = iterator.next()
val tmp = DBOps(key)
if(tmp != null){
if((System.currentTimeMillis() - tmp.getUseTime) > this.checktime){
exists += key
}
}
}
exists.foreach(f => {
try{
DBOps -= f
}catch{
case e:Exception =>{
log.error("MultiDBMgr_" + SystemName + "_" + s"ImpDataPreload clear failed, error info= "+e.getMessage)
throw e
}
}
})
}
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* @category 释放指定的实例
* @param InstanceName String 实例名称
* @return 无
* */
def Free(InstanceName:String)={
try{
//DBOps.remove(InstanceName)
DBOps -= InstanceName
}catch{
case e:Exception =>{
log.error("MultiDBMgr_" + SystemName + "_" + s"ImpDataPreload Free failed, error info= "+e.getMessage)
throw e
}
}
}
}
////////////////以下半生对象用来做存储实例的管理，相当于存储管理的类工厂，每一个系统只能产生一个实例来进行操作///////////////
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* @category 系统管理类，管理每个系统中的多实例管理器。
* */
object ImpDataPreloadMgr{
import org.slf4j.LoggerFactory
protected def log = LoggerFactory.getLogger(this.getClass)
private var singleobjs = new scala.collection.mutable.HashMap[String,MultiDBMgr]()
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* @category 根据系统名称、实例名称获取预执行的数据访问类
* @param SystemName String 系统名称,InstanceName String 实例名称
* @return 如果成功返回ImpDataPreload实例，否则为null
* */
def GetImpDataPreload( SystemName:String,InstanceName:String) : ImpDataPreload = {
var singleobj : MultiDBMgr = null
var dbop : ImpDataPreload = null
synchronized{
try{
//clear //暂时先不每次做全局清除
if(singleobjs.contains(SystemName)){
singleobj = singleobjs(SystemName)
}else{
singleobj = new MultiDBMgr(SystemName)
singleobjs.put(SystemName, singleobj)
}
dbop = singleobj.GetImpDataPreload(InstanceName)
}catch{
case e:Exception =>{
log.error("ImpDataPreloadMgr_" + SystemName + "_" + s"ImpDataPreload Create failed, error info= "+e.getMessage)
throw e
}
}
dbop
}
}
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* @category 从多实例管理器中清理系统中空闲超时的实例
* @param 无
* @return 无
* */
private def clear{
val iterator = singleobjs.keysIterator
while (iterator.hasNext) {
val key = iterator.next()
val tmp = singleobjs(key)
if(tmp != null){
tmp.clear
}
}
}
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* @category 从多实例管理器中释放指定的实例
* @param SystemName String 系统名称，InstanceName String 实例名称
* @return 无
* */
def Free(SystemName:String,InstanceName:String)={
var singleobj : MultiDBMgr = null
synchronized{
try{
if(singleobjs.contains(SystemName)){
singleobj = singleobjs(SystemName)
if(singleobj != null) singleobj.Free(InstanceName)
}
}catch{
case e:Exception =>{
log.error("ImpDataPreloadMgr_" + SystemName + "_" + s"ImpDataPreload Free failed, error info= "+e.getMessage)
throw e
}
}
}
}
}
/*
* Copyright 2018 Blockchain Technology and Application Joint Lab, Linkel Technology Co., Ltd, Beijing, Fintech Research Center of ISCAS.
* 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
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BA SIS,
* 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 rep.storage
import rep.utils._
import scala.collection.mutable
import rep.storage.leveldb._
import rep.storage.merkle._
import scala.collection.mutable.ArrayBuffer
import rep.protos.peer._;
import scala.util.control.Breaks
import rep.log.trace._
/**内存数据库的访问类，属于多实例。
* @constructor 根据SystemName和InstanceName建立实例
* @param SystemName 系统名
* @param InstanceName 实例名
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* @category 内存数据库的访问类，属于多实例。
*/
class ImpDataPreload (SystemName:String,InstanceName:String) extends AbstractLevelDB(SystemName:String) {
private var update :java.util.concurrent.ConcurrentHashMap[String,Array[Byte]] = new java.util.concurrent.ConcurrentHashMap[String,Array[Byte]]
private var dbop = ImpDataAccess.GetDataAccess(SystemName)
//private var merkleop : RepBucket = new RepBucket(this)
this.ReloadMerkle
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* @category 获取当前系统的名称
* @param 无
* @return 返回当前系统的名称 String
* */
override def getSystemName:String={
SystemName
}
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* @category 获取当前实例的名称
* @param 无
* @return 返回当前实例的名称 String
* */
override def getInstanceName:String={
InstanceName
}
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* @category 获取指定的键值
* @param key String 指定的键
* @return 返回对应键的值 Array[Byte]
* */
override def Get(key : String):Array[Byte]={
var rb : Array[Byte] = null
try{
if(this.update.containsKey(key)){
rb = this.update.get(key)
}else{
rb = this.dbop.Get(key)
}
setUseTime
}catch{
case e:Exception =>{
rb = null
RepLogger.logError(SystemName, ModuleType.storager,
"ImpDataPreload_" + SystemName + "_" + s"ImpDataPreload Get failed, error info= "+e.getMessage)
throw e
}
}
rb
}
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* @category 存储指定的键和值到数据库
* @param key String 指定的键，bb Array[Byte] 要存储的值
* @return 返回成功或者失败 Boolean
* */
override def Put (key : String,bb : Array[Byte],isWorldState : Boolean):Boolean={
var b : Boolean = true
try{
if(key == null){
RepLogger.logInfo(SystemName, ModuleType.storager,
"ImpDataPreload_" + SystemName + "_" + s"ImpDataPreload Put failed, error info= key is null")
}
if(bb == null){
RepLogger.logInfo(SystemName, ModuleType.storager,
"ImpDataPreload_" + SystemName + "_" + s"ImpDataPreload Put failed, error info= value is null")
}
if(key != null && bb != null){
if(this.update.get(key) != null){
val obbstr = toString(this.update.get(key))
val bbstr = toString(bb)
if(!obbstr.equals(bbstr)){
this.update.put(key, bb)
//this.merkleop.Put(key, bb)
this.PutWorldStateToMerkle(key,bb)
}
}else{
if(this.dbop.Get(key) != null){
val obbstr = toString(this.dbop.Get(key))
val bbstr = toString(bb)
if(!obbstr.equals(bbstr)){
this.update.put(key, bb)
//this.merkleop.Put(key, bb)
this.PutWorldStateToMerkle(key,bb)
}
}else{
this.update.put(key, bb)
//this.merkleop.Put(key, bb)
this.PutWorldStateToMerkle(key,bb)
}
}
}
setUseTime
}catch{
case e:Exception =>{
b = false
RepLogger.logError(SystemName, ModuleType.storager,
"ImpDataPreload_" + SystemName + "_" + s"ImpDataPreload Put failed, error info= "+e.getMessage)
throw e
}
}
b
}
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* @category 删除指定的键值
* @param key String 指定的键
* @return 返回成功或者失败 Boolean
* 该类暂时没有实现，因为RepChain不能够删除已有的WorldState
* */
override def Delete (key : String) : Boolean={
var b : Boolean = true;
//todo 目前没有实现，原则上不允许删除任何状态
setUseTime
b;
}
private var useTime:Long = System.currentTimeMillis();
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* @category 内部使用，更新实例的访问时间，每次访问的时候都会调用该方法
* @param key String 指定的键
* @return 无
* */
def setUseTime{
this.useTime = System.currentTimeMillis();
}
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* @category 内部使用，获取该实例最后一次使用时间
* @param 无
* @return 长整型，最后一次的使用时间
* */
def getUseTime:Long={
this.useTime
}
//////////////////////////////Endorsement 背书验证/////////////////////////////////////////////////////////////////////////////
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* @category 根据交易id获取chaincode id
* @param block Block 待写入的区块,txid String 待查询的交易id
* @return 如果成功返回chaincode id，否则返回空字符串
* */
private def getTxidFormBlock(block:Block,txid:String):String={
var rel = ""
if(block != null){
var trans = block.transactions
if(trans.length > 0){
val loopbreak = new Breaks
loopbreak.breakable(
trans.foreach(f=>{
if(f.id.equals(txid)){
rel = this.getCid(f.cid.get)
//val chainspec = f.payload.get
//rel = chainspec.chaincodeID.get.name
loopbreak.break
}
})
)
}
}
rel
}
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* @category 交易背书时调用，用于验证读写指令集
* @param block Block 待验证的区块
* @return 如果验证成功返回true，否则false
* */
def VerifyForEndorsement(block:Block):Boolean={
var b : Boolean = false
if(block == null) return b
if(block.operHash == null) return b
try{
var list : scala.collection.mutable.LinkedHashMap[String,Seq[OperLog]] = new scala.collection.mutable.LinkedHashMap[String,Seq[OperLog]]()
val txresults = block.transactionResults
txresults.foreach(f=>{
list += f.txId -> f.ol
}
)
if(list.size > 0){
list.foreach(f=>{
val txid = f._1
val cid = getTxidFormBlock(block,txid)
val jobj = f._2
if(jobj != null && jobj.length > 0){
jobj.foreach(f=>{
var tmpkeystr = IdxPrefix.WorldStateKeyPreFix+cid+"_"+f.key
this.Put(IdxPrefix.WorldStateKeyPreFix+cid+"_"+f.key, f.newValue.toByteArray())
})
}
})
}
val shash4block = block.operHash.toStringUtf8
val shash4local = this.GetComputeMerkle4String
if(shash4block.equals(shash4local)){
b = true
}else{
b = false
RepLogger.logError(SystemName, ModuleType.storager,
"system_name="+this.SystemName+"\t verify World State is failed, hash4block="+shash4block+"\t hash4local="+shash4local)
}
}catch{
case e:Exception =>{
throw e;
}
}finally{
//todo
}
b
}
/////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
////////////////////////////////worldstate merkle 继承实现///////////////////////////////////////////////////////////////////////
/*override def GetComputeMerkle:Array[Byte]={
setUseTime
val b = this.merkleop.getMerkleHash
b
}
override def GetComputeMerkle4String:String={
setUseTime
val c = this.merkleop.getMerkelHash4String
c
}*/
///////////////////////////////////////////////////////////////////////////////////////////////////////
}
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* @category 某个系统的多实例管理类，管理某个系统所有的多实例，包含自动释放超时的实例。
* */
private class MultiDBMgr (val SystemName:String) {
import org.slf4j.LoggerFactory
private val checktime = 60*1000//如果某个实例超过60s没有使用，就自动清理
private var DBOps = new scala.collection.mutable.HashMap[String,ImpDataPreload]()
protected def log = LoggerFactory.getLogger(this.getClass)
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* @category 根据实例名称获取预执行的数据访问类
* @param InstanceName String 实例名称
* @return 如果成功返回ImpDataPreload实例，否则为null
* */
def GetImpDataPreload( InstanceName:String) : ImpDataPreload = {
var DBOp : ImpDataPreload = null
synchronized{
try{
clear
if(DBOps.contains(InstanceName)){
DBOp = DBOps(InstanceName)
}else{
DBOp = new ImpDataPreload(SystemName,InstanceName)
DBOps.put(InstanceName, DBOp)
}
if(DBOp != null){
DBOp.setUseTime
}
}catch{
case e:Exception =>{
RepLogger.logError(SystemName, ModuleType.storager,
"MultiDBMgr_" + SystemName + "_" + s"ImpDataPreload Create failed, error info= "+e.getMessage)
throw e
}
}
DBOp
}
}
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* @category 清理系统中空闲超时的实例
* @param 无
* @return 无
* */
def clear{
var exists : ArrayBuffer[String] = new ArrayBuffer[String]()
val iterator = DBOps.keysIterator
while (iterator.hasNext) {
val key = iterator.next()
val tmp = DBOps(key)
if(tmp != null){
if((System.currentTimeMillis() - tmp.getUseTime) > this.checktime){
exists += key
}
}
}
exists.foreach(f => {
try{
DBOps -= f
}catch{
case e:Exception =>{
RepLogger.logError(SystemName, ModuleType.storager,
"MultiDBMgr_" + SystemName + "_" + s"ImpDataPreload clear failed, error info= "+e.getMessage)
throw e
}
}
})
}
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* @category 释放指定的实例
* @param InstanceName String 实例名称
* @return 无
* */
def Free(InstanceName:String)={
try{
//DBOps.remove(InstanceName)
DBOps -= InstanceName
}catch{
case e:Exception =>{
RepLogger.logError(SystemName, ModuleType.storager,
"MultiDBMgr_" + SystemName + "_" + s"ImpDataPreload Free failed, error info= "+e.getMessage)
throw e
}
}
}
}
////////////////以下半生对象用来做存储实例的管理，相当于存储管理的类工厂，每一个系统只能产生一个实例来进行操作///////////////
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* @category 系统管理类，管理每个系统中的多实例管理器。
* */
object ImpDataPreloadMgr{
import org.slf4j.LoggerFactory
protected def log = LoggerFactory.getLogger(this.getClass)
private var singleobjs = new scala.collection.mutable.HashMap[String,MultiDBMgr]()
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* @category 根据系统名称、实例名称获取预执行的数据访问类
* @param SystemName String 系统名称,InstanceName String 实例名称
* @return 如果成功返回ImpDataPreload实例，否则为null
* */
def GetImpDataPreload( SystemName:String,InstanceName:String) : ImpDataPreload = {
var singleobj : MultiDBMgr = null
var dbop : ImpDataPreload = null
synchronized{
try{
//clear //暂时先不每次做全局清除
if(singleobjs.contains(SystemName)){
singleobj = singleobjs(SystemName)
}else{
singleobj = new MultiDBMgr(SystemName)
singleobjs.put(SystemName, singleobj)
}
dbop = singleobj.GetImpDataPreload(InstanceName)
}catch{
case e:Exception =>{
RepLogger.logError(SystemName, ModuleType.storager,
"ImpDataPreloadMgr_" + SystemName + "_" + s"ImpDataPreload Create failed, error info= "+e.getMessage)
throw e
}
}
dbop
}
}
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* @category 从多实例管理器中清理系统中空闲超时的实例
* @param 无
* @return 无
* */
private def clear{
val iterator = singleobjs.keysIterator
while (iterator.hasNext) {
val key = iterator.next()
val tmp = singleobjs(key)
if(tmp != null){
tmp.clear
}
}
}
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* @category 从多实例管理器中释放指定的实例
* @param SystemName String 系统名称，InstanceName String 实例名称
* @return 无
* */
def Free(SystemName:String,InstanceName:String)={
var singleobj : MultiDBMgr = null
synchronized{
try{
if(singleobjs.contains(SystemName)){
singleobj = singleobjs(SystemName)
if(singleobj != null) singleobj.Free(InstanceName)
}
}catch{
case e:Exception =>{
RepLogger.logError(SystemName, ModuleType.storager,
"ImpDataPreloadMgr_" + SystemName + "_" + s"ImpDataPreload Free failed, error info= "+e.getMessage)
throw e
}
}
}
}
}
////////////////end///////////////

416
src/main/scala/rep/storage/blockindex.scala
View File

@ -1,209 +1,209 @@
/*
* Copyright 2018 Blockchain Technology and Application Joint Lab, Linkel Technology Co., Ltd, Beijing, Fintech Research Center of ISCAS.
* 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
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BA SIS,
* 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 rep.storage
import scala.util.parsing.json._;
import rep.protos.peer._;
import rep.storage.leveldb._;
import rep.storage.cfg._;
import rep.crypto._;
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* */
class blockindex() {
private var blockNum : String = "repchain_default";//系统中暂时没有这个内容
private var blockHeight : Long = -1;
private var blockHash : String = "";
private var blockPrevHash : String = "";
private var txids : Array[String] = null;
private var BlockFileNo : Int = 0;
private var BlockFilePos : Long = 0;
private var BlockLength : Int = 0;
def InitBlockIndex(b : Block)={
if(b != null){
val rbb = b.toByteArray
this.blockHash = Sha256.hashstr(rbb);
//不能使用statehash作为block的hashid
//this.blockHash = b.stateHash.toString("UTF-8");
this.blockPrevHash = b.previousBlockHash.toString("UTF-8");
val ts = b.transactions;
if(ts != null && ts.length > 0){
txids = new Array[String](ts.length);
var i = 0;
ts.foreach(f =>{
txids(i) = f.txid;
i += 1;
})
}
}
}
def InitBlockIndex(ab : Array[Byte])={
if(ab != null){
val jstr = new String(ab,"UTF-8");
if(jstr != null){
var m:Map[String,Any] = JsonUtil.json2Map(jstr);
this.blockNum = this.str4null(getAnyType(m,"blockNum"));
this.blockHeight = str2Long(getAnyType(m,"blockHeight"));
this.blockHash = this.str4null(getAnyType(m,"blockHash"));;
this.blockPrevHash = this.str4null(getAnyType(m,"blockPrevHash"));;
this.txids = this.str2Array(getAnyType(m,"txids"));
this.BlockFileNo = str2Int(getAnyType(m,"BlockFileNo"));
this.BlockFilePos = str2Long(getAnyType(m,"BlockFilePos"));;
this.BlockLength = str2Int(getAnyType(m,"BlockLength"));;
}
}
}
def getAnyType(m:Map[String,Any],key:String):String={
var rstr = "";
if(m.contains(key)) {
val mv = m(key);
if(mv == null){
rstr = "";
}else{
rstr = mv.toString();
}
}
rstr;
}
def str2Int(str : String):Int={
var ri = 0;
if(str != null){
if(!str.equalsIgnoreCase("") && !str.equalsIgnoreCase("null")){
ri = Integer.parseInt(str);
}
}
ri;
}
def str2Long(str : String):Long={
var rl:Long = 0;
if(str != null){
if(!str.equalsIgnoreCase("") && !str.equalsIgnoreCase("null")){
rl = java.lang.Long.parseLong(str);
}
}
rl;
}
def str4null(str : String):String={
var rs = "";
if(str == null){
rs = "";
}else if(str.equalsIgnoreCase("null")){
rs = "";
}else{
rs = str;
}
rs;
}
def str2Array(str : String):Array[String]={
var ra : Array[String] = null;
if(str != null && !str.equalsIgnoreCase("") && !str.equalsIgnoreCase("null")){
ra = str.split(" ");
}
ra;
}
def toJsonStr():String={
var rstr = "";
val map = scala.collection.mutable.HashMap[String,Any]();
map.put("blockNum", blockNum );
map.put("blockHeight", String.valueOf(blockHeight));
map.put("blockHash", blockHash);
map.put("blockPrevHash", blockPrevHash);
if(txids != null && txids.length > 0){
val str = txids.mkString(" ");
map.put("txids", str);
}
map.put("BlockFileNo", String.valueOf(BlockFileNo));
map.put("BlockFilePos", String.valueOf(BlockFilePos));
map.put("BlockLength", String.valueOf(BlockLength));
rstr = JsonUtil.map2Json(map.toMap);
rstr;
}
def toArrayByte():Array[Byte]={
val rstr = toJsonStr();
val rb = rstr.getBytes("UTF-8");
rb;
}
def getBlockNum():String={
this.blockNum;
}
def setBlockNum(s:String)={
this.blockNum = s;
}
def getBlockHeight():Long={
this.blockHeight;
}
def setBlockHeight(l:Long)={
this.blockHeight = l;
}
def getBlockHash():String={
this.blockHash;
}
def getBlockPrevHash():String={
this.blockPrevHash;
}
def getTxIds():Array[String]={
this.txids;
}
def getBlockFileNo():Int={
this.BlockFileNo;
}
def setBlockFileNo(l:Int)={
this.BlockFileNo = l;
}
def getBlockFilePos():Long={
this.BlockFilePos;
}
def setBlockFilePos(l:Long)={
this.BlockFilePos = l;
}
def getBlockLength():Int={
this.BlockLength;
}
def setBlockLength(l:Int)={
this.BlockLength = l;
}
/*
* Copyright 2018 Blockchain Technology and Application Joint Lab, Linkel Technology Co., Ltd, Beijing, Fintech Research Center of ISCAS.
* 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
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BA SIS,
* 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 rep.storage
import scala.util.parsing.json._;
import rep.protos.peer._;
import rep.storage.leveldb._;
import rep.storage.cfg._;
import rep.crypto._;
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* */
class blockindex() {
private var blockNum : String = "repchain_default";//系统中暂时没有这个内容
private var blockHeight : Long = -1;
private var blockHash : String = "";
private var blockPrevHash : String = "";
private var txids : Array[String] = null;
private var BlockFileNo : Int = 0;
private var BlockFilePos : Long = 0;
private var BlockLength : Int = 0;
def InitBlockIndex(b : Block)={
if(b != null){
val rbb = b.toByteArray
this.blockHash = Sha256.hashstr(rbb);
//不能使用statehash作为block的hashid
//this.blockHash = b.stateHash.toString("UTF-8");
this.blockPrevHash = b.previousBlockHash.toString("UTF-8");
val ts = b.transactions;
if(ts != null && ts.length > 0){
txids = new Array[String](ts.length);
var i = 0;
ts.foreach(f =>{
txids(i) = f.id;
i += 1;
})
}
}
}
def InitBlockIndex(ab : Array[Byte])={
if(ab != null){
val jstr = new String(ab,"UTF-8");
if(jstr != null){
var m:Map[String,Any] = JsonUtil.json2Map(jstr);
this.blockNum = this.str4null(getAnyType(m,"blockNum"));
this.blockHeight = str2Long(getAnyType(m,"blockHeight"));
this.blockHash = this.str4null(getAnyType(m,"blockHash"));;
this.blockPrevHash = this.str4null(getAnyType(m,"blockPrevHash"));;
this.txids = this.str2Array(getAnyType(m,"txids"));
this.BlockFileNo = str2Int(getAnyType(m,"BlockFileNo"));
this.BlockFilePos = str2Long(getAnyType(m,"BlockFilePos"));;
this.BlockLength = str2Int(getAnyType(m,"BlockLength"));;
}
}
}
def getAnyType(m:Map[String,Any],key:String):String={
var rstr = "";
if(m.contains(key)) {
val mv = m(key);
if(mv == null){
rstr = "";
}else{
rstr = mv.toString();
}
}
rstr;
}
def str2Int(str : String):Int={
var ri = 0;
if(str != null){
if(!str.equalsIgnoreCase("") && !str.equalsIgnoreCase("null")){
ri = Integer.parseInt(str);
}
}
ri;
}
def str2Long(str : String):Long={
var rl:Long = 0;
if(str != null){
if(!str.equalsIgnoreCase("") && !str.equalsIgnoreCase("null")){
rl = java.lang.Long.parseLong(str);
}
}
rl;
}
def str4null(str : String):String={
var rs = "";
if(str == null){
rs = "";
}else if(str.equalsIgnoreCase("null")){
rs = "";
}else{
rs = str;
}
rs;
}
def str2Array(str : String):Array[String]={
var ra : Array[String] = null;
if(str != null && !str.equalsIgnoreCase("") && !str.equalsIgnoreCase("null")){
ra = str.split(" ");
}
ra;
}
def toJsonStr():String={
var rstr = "";
val map = scala.collection.mutable.HashMap[String,Any]();
map.put("blockNum", blockNum );
map.put("blockHeight", String.valueOf(blockHeight));
map.put("blockHash", blockHash);
map.put("blockPrevHash", blockPrevHash);
if(txids != null && txids.length > 0){
val str = txids.mkString(" ");
map.put("txids", str);
}
map.put("BlockFileNo", String.valueOf(BlockFileNo));
map.put("BlockFilePos", String.valueOf(BlockFilePos));
map.put("BlockLength", String.valueOf(BlockLength));
rstr = JsonUtil.map2Json(map.toMap);
rstr;
}
def toArrayByte():Array[Byte]={
val rstr = toJsonStr();
val rb = rstr.getBytes("UTF-8");
rb;
}
def getBlockNum():String={
this.blockNum;
}
def setBlockNum(s:String)={
this.blockNum = s;
}
def getBlockHeight():Long={
this.blockHeight;
}
def setBlockHeight(l:Long)={
this.blockHeight = l;
}
def getBlockHash():String={
this.blockHash;
}
def getBlockPrevHash():String={
this.blockPrevHash;
}
def getTxIds():Array[String]={
this.txids;
}
def getBlockFileNo():Int={
this.BlockFileNo;
}
def setBlockFileNo(l:Int)={
this.BlockFileNo = l;
}
def getBlockFilePos():Long={
this.BlockFilePos;
}
def setBlockFilePos(l:Long)={
this.BlockFilePos = l;
}
def getBlockLength():Int={
this.BlockLength;
}
def setBlockLength(l:Int)={
this.BlockLength = l;
}
}

462
src/main/scala/rep/storage/leveldb/AbstractLevelDB.scala
View File

@ -1,228 +1,236 @@
/*
* Copyright 2018 Blockchain Technology and Application Joint Lab, Linkel Technology Co., Ltd, Beijing, Fintech Research Center of ISCAS.
* 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
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BA SIS,
* 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 rep.storage.leveldb
import scala.collection.immutable
import scala.collection.mutable
import rep.utils._
import scala.collection.JavaConversions._
import scala.collection.JavaConverters._
import rep.storage.IdxPrefix
import rep.storage.util.StoreUtil
import com.google.protobuf.ByteString
import scala.math._
import rep.crypto._
import rep.log.trace.RepLogHelp
import rep.log.trace.LogType
import org.slf4j.LoggerFactory
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* @category 该类实现公共方法。
* */
abstract class AbstractLevelDB extends ILevelDB {
protected def log = LoggerFactory.getLogger(this.getClass)
protected var IncrementWorldState : immutable.TreeMap[String,Array[Byte]] = new immutable.TreeMap[String,Array[Byte]]()
protected var GlobalMerkle : Array[Byte] = null
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* @category 初始化Merkle值
* @param 无
* @return 无
* */
protected def ReloadMerkle={
this.GlobalMerkle = null
this.IncrementWorldState = new immutable.TreeMap[String,Array[Byte]]()
val key = IdxPrefix.WorldStateForInternetPrefix + IdxPrefix.GlobalWorldStateValue
val v = this.Get(key)
if(v != null){
this.GlobalMerkle = v
}
}
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* @category 系统WorldState改变时需要调用该函数完成WorldState的Merkle重新计算
* @param key 指定的键，value Array[Byte] 修改的键值
* @return 无
* */
protected def PutWorldStateToMerkle(key:String,value:Array[Byte]){
val prefix = IdxPrefix.WorldStateKeyPreFix
if(key.startsWith(prefix)){
val sv = Sha256.hash(value)
this.IncrementWorldState += key -> sv
}
}
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* @category 计算当前WorldState的Merkle的值
* @param 无
* @return 返回WorldState的Merkle值 Array[Byte]
* */
override def GetComputeMerkle:Array[Byte]={
var rel :Array[Byte] = null
val source = this.IncrementWorldState.values.toArray
if(source.size > 0){
var value : Array[Byte] = null
var i = 1
if(this.GlobalMerkle != null){
i = 0
value = this.GlobalMerkle
}else{
value = source(0)
i = 1
}
while(i < source.size){
value = Array.concat(value , source(i))
i += 1
}
rel = Sha256.hash(value)
}else{
rel = this.GlobalMerkle
}
if(rel != null){
//println("=========################getmerkle value="+BytesHex.bytes2hex(rel)+"\tsource size="+source.size)
}else{
//println("=========################getmerkle value=null"+"\tsource size="+source.size)
}
rel
}
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* @category 计算当前WorldState的Merkle的值
* @param 无
* @return 返回WorldState的Merkle值 String
* */
override def GetComputeMerkle4String:String={
var rel:String = ""
val bb = GetComputeMerkle
if(bb != null){
rel = BytesHex.bytes2hex(bb)
}
rel
}
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* @category 把字节数组转成字符串
* @param b Array[Byte] 待转换字节数组
* @return 返回转换结果，String 如果为null 返回空字符串
* */
def toString(b : Array[Byte]):String={
var str : String = ""
if(b != null){
str = new String(b)
}
str
}
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* @category 把字节数组转成长整型
* @param b Array[Byte] 待转换字节数组
* @return 返回转换结果，Long 如果为null 返回-1
* */
def toLong(b : Array[Byte]):Long={
var l : Long = -1
if(b != null){
val str = toString(b)
try{
l = str.toLong
}catch{
case e:Exception =>{
log.error( s"DBOP toLong failed, error info= "+e.getMessage)
}
}
}
l
}
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* @category 把字节数组转成整型
* @param b Array[Byte] 待转换字节数组
* @return 返回转换结果，Int 如果为null 返回-1
* */
def toInt(b : Array[Byte]):Int={
var l : Int = -1
if(b != null){
val str = toString(b)
try{
l = str.toInt
}catch{
case e:Exception =>{
log.error( s"DBOP toInt failed, error info= "+e.getMessage)
}
}
}
l
}
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* @category 把字节数组转成字符串
* @param a Array[Byte] 待转换的字节数组
* @return 返回转换结果，String 如果为null 返回空字符串
* */
def byteToString(a:Array[Byte]):String={
var s = ""
if(a != null){
s = new String(a)
}
s
}
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* @category 打印Map中的键值对
* @param map 需要打印的map
* @return 无
* */
def printlnHashMap(map : mutable.HashMap[String,Array[Byte]])={
if(map != null){
map.foreach(f=>{
log.warn("\tkey="+f._1 + "\tvalue=" +toString(f._2))
})
}
}
/*
* Copyright 2018 Blockchain Technology and Application Joint Lab, Linkel Technology Co., Ltd, Beijing, Fintech Research Center of ISCAS.
* 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
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BA SIS,
* 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 rep.storage.leveldb
import scala.collection.immutable
import scala.collection.mutable
import rep.utils._
import scala.collection.JavaConversions._
import scala.collection.JavaConverters._
import rep.storage.IdxPrefix
import rep.storage.util.StoreUtil
import com.google.protobuf.ByteString
import scala.math._
import rep.crypto._
import rep.log.trace.LogType
import org.slf4j.LoggerFactory
import rep.log.trace._
import rep.protos.peer.ChaincodeID
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* @category 该类实现公共方法。
* */
abstract class AbstractLevelDB(SystemName:String) extends ILevelDB {
//protected def log = LoggerFactory.getLogger(this.getClass)
protected var IncrementWorldState : immutable.TreeMap[String,Array[Byte]] = new immutable.TreeMap[String,Array[Byte]]()
protected var GlobalMerkle : Array[Byte] = null
protected def getCid(chaincodeid:ChaincodeID):String={
chaincodeid.chaincodeName+"_"+chaincodeid.version.toString()
}
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* @category 初始化Merkle值
* @param 无
* @return 无
* */
protected def ReloadMerkle={
this.GlobalMerkle = null
this.IncrementWorldState = new immutable.TreeMap[String,Array[Byte]]()
val key = IdxPrefix.WorldStateForInternetPrefix + IdxPrefix.GlobalWorldStateValue
val v = this.Get(key)
if(v != null){
this.GlobalMerkle = v
}
}
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* @category 系统WorldState改变时需要调用该函数完成WorldState的Merkle重新计算
* @param key 指定的键，value Array[Byte] 修改的键值
* @return 无
* */
protected def PutWorldStateToMerkle(key:String,value:Array[Byte]){
val prefix = IdxPrefix.WorldStateKeyPreFix
if(key.startsWith(prefix)){
val sv = Sha256.hash(value)
this.IncrementWorldState += key -> sv
}
}
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* @category 计算当前WorldState的Merkle的值
* @param 无
* @return 返回WorldState的Merkle值 Array[Byte]
* */
override def GetComputeMerkle:Array[Byte]={
var rel :Array[Byte] = null
val source = this.IncrementWorldState.values.toArray
if(source.size > 0){
var value : Array[Byte] = null
var i = 1
if(this.GlobalMerkle != null){
i = 0
value = this.GlobalMerkle
}else{
value = source(0)
i = 1
}
while(i < source.size){
value = Array.concat(value , source(i))
i += 1
}
rel = Sha256.hash(value)
}else{
rel = this.GlobalMerkle
}
if(rel != null){
//println("=========################getmerkle value="+BytesHex.bytes2hex(rel)+"\tsource size="+source.size)
}else{
//println("=========################getmerkle value=null"+"\tsource size="+source.size)
}
rel
}
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* @category 计算当前WorldState的Merkle的值
* @param 无
* @return 返回WorldState的Merkle值 String
* */
override def GetComputeMerkle4String:String={
var rel:String = ""
val bb = GetComputeMerkle
if(bb != null){
rel = BytesHex.bytes2hex(bb)
}
rel
}
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* @category 把字节数组转成字符串
* @param b Array[Byte] 待转换字节数组
* @return 返回转换结果，String 如果为null 返回空字符串
* */
def toString(b : Array[Byte]):String={
var str : String = ""
if(b != null){
str = new String(b)
}
str
}
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* @category 把字节数组转成长整型
* @param b Array[Byte] 待转换字节数组
* @return 返回转换结果，Long 如果为null 返回-1
* */
def toLong(b : Array[Byte]):Long={
var l : Long = -1
if(b != null){
val str = toString(b)
try{
l = str.toLong
}catch{
case e:Exception =>{
RepLogger.logError(SystemName, ModuleType.storager,
s"DBOP toLong failed, error info= "+e.getMessage)
}
}
}
l
}
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* @category 把字节数组转成整型
* @param b Array[Byte] 待转换字节数组
* @return 返回转换结果，Int 如果为null 返回-1
* */
def toInt(b : Array[Byte]):Int={
var l : Int = -1
if(b != null){
val str = toString(b)
try{
l = str.toInt
}catch{
case e:Exception =>{
RepLogger.logError(SystemName, ModuleType.storager,
s"DBOP toInt failed, error info= "+e.getMessage)
}
}
}
l
}
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* @category 把字节数组转成字符串
* @param a Array[Byte] 待转换的字节数组
* @return 返回转换结果，String 如果为null 返回空字符串
* */
def byteToString(a:Array[Byte]):String={
var s = ""
if(a != null){
s = new String(a)
}
s
}
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* @category 打印Map中的键值对
* @param map 需要打印的map
* @return 无
* */
def printlnHashMap(map : mutable.HashMap[String,Array[Byte]])={
if(map != null){
map.foreach(f=>{
RepLogger.logWarn(SystemName, ModuleType.storager,
"\tkey="+f._1 + "\tvalue=" +toString(f._2))
})
}
}
}

293
src/main/scala/rep/storage/merkle/RepBucket.scala
View File

@ -1,148 +1,147 @@
/*
* Copyright 2018 Blockchain Technology and Application Joint Lab, Linkel Technology Co., Ltd, Beijing, Fintech Research Center of ISCAS.
* 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
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BA SIS,
* 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 rep.storage.merkle
import rep.storage.leveldb.ILevelDB
import scala.collection.immutable
import scala.collection.mutable
import rep.storage.IdxPrefix
import rep.utils._
import rep.crypto._
import rep.storage.util.StoreUtil
import rep.log.trace.RepLogHelp
import rep.log.trace.LogType
import org.slf4j.LoggerFactory
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* */
class RepBucket(val leveldb : ILevelDB){
protected def log = LoggerFactory.getLogger(this.getClass)
private var ccids : mutable.ArrayBuffer[String] = new mutable.ArrayBuffer[String]()
private var MerkleHash : Array[Byte] = null
private var ccbucket : mutable.HashMap[String,ChainCodeBucket] = new mutable.HashMap[String,ChainCodeBucket]()
load
private def load={
getMerkleHashFromDB
getAllChainCode
this.ccids.foreach(f=>{
if(!this.ccbucket.contains(f)){
var tmp = new ChainCodeBucket(this.leveldb,f)
this.ccbucket += f -> tmp
}
})
}
def Put( key:String, value:Array[Byte]){
synchronized{
if(key.startsWith(IdxPrefix.WorldStateKeyPreFix)){
val keys = StoreUtil.SplitKey(key)
if(keys != null && keys.length == 3){
val ccid = keys(1)
if(this.ccbucket.contains(ccid)){
var tmp = this.ccbucket(ccid)
if(tmp != null){
tmp.Put(key, value)
computeMerkel
}
}else{
this.ccids += ccid
var tmp1 = new ChainCodeBucket(this.leveldb,ccid)
this.ccbucket += ccid -> tmp1
tmp1.Put(key, value)
computeMerkel
}
}
}
}
}
private def computeMerkel={
synchronized{
val start = System.currentTimeMillis()
var value : Array[Byte] = null
this.ccids.foreach(f=>{
var tmp = this.ccbucket(f)
if(tmp != null){
val tv = tmp.getCCMerkle
if(tv != null){
if(value == null){
value = tv
}else{
value = Array.concat(value , tv)
}
}
}
})
if(value != null) this.MerkleHash = Sha256.hash(value)
val end = System.currentTimeMillis()
log.info("Merkle compute time="+(end - start)+"ms")
}
}
def getMerkleHash:Array[Byte]={
this.MerkleHash
}
def getMerkelHash4String:String={
var rel:String = ""
val bb = getMerkleHash
if(bb != null){
rel = BytesHex.bytes2hex(bb)
}
rel
}
def save={
val keyhash = IdxPrefix.WorldStateForInternetPrefix + "merkletreehash"
this.leveldb.Put(keyhash, this.MerkleHash)
val key = IdxPrefix.WorldStateForInternetPrefix + IdxPrefix.ChainCodeList
val cckb = SerializeUtils.serialise(this.ccids)
this.leveldb.Put(key, cckb)
this.ccids.foreach(f=>{
var tmp = this.ccbucket(f)
if(tmp != null){
tmp.save
}
})
}
private def getMerkleHashFromDB={
val key = IdxPrefix.WorldStateForInternetPrefix + "merkletreehash"
val v = leveldb.Get(key)
if(v != null){
this.MerkleHash = v
}
}
private def getAllChainCode={
val key = IdxPrefix.WorldStateForInternetPrefix + IdxPrefix.ChainCodeList
val v = leveldb.Get(key)
if(v != null){
val tmp = SerializeUtils.deserialise(v)
if(tmp.isInstanceOf[mutable.ArrayBuffer[String]] ){
var ta = tmp.asInstanceOf[mutable.ArrayBuffer[String]]
this.ccids = ta
}
}
}
/*
* Copyright 2018 Blockchain Technology and Application Joint Lab, Linkel Technology Co., Ltd, Beijing, Fintech Research Center of ISCAS.
* 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
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BA SIS,
* 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 rep.storage.merkle
import rep.storage.leveldb.ILevelDB
import scala.collection.immutable
import scala.collection.mutable
import rep.storage.IdxPrefix
import rep.utils._
import rep.crypto._
import rep.storage.util.StoreUtil
import rep.log.trace.LogType
import org.slf4j.LoggerFactory
/**
* @author jiangbuyun
* @version 0.7
* @since 2017-09-28
* */
class RepBucket(val leveldb : ILevelDB){
protected def log = LoggerFactory.getLogger(this.getClass)
private var ccids : mutable.ArrayBuffer[String] = new mutable.ArrayBuffer[String]()
private var MerkleHash : Array[Byte] = null
private var ccbucket : mutable.HashMap[String,ChainCodeBucket] = new mutable.HashMap[String,ChainCodeBucket]()
load
private def load={
getMerkleHashFromDB
getAllChainCode
this.ccids.foreach(f=>{
if(!this.ccbucket.contains(f)){
var tmp = new ChainCodeBucket(this.leveldb,f)
this.ccbucket += f -> tmp
}
})
}
def Put( key:String, value:Array[Byte]){
synchronized{
if(key.startsWith(IdxPrefix.WorldStateKeyPreFix)){
val keys = StoreUtil.SplitKey(key)
if(keys != null && keys.length == 3){
val ccid = keys(1)
if(this.ccbucket.contains(ccid)){
var tmp = this.ccbucket(ccid)
if(tmp != null){
tmp.Put(key, value)
computeMerkel
}
}else{
this.ccids += ccid
var tmp1 = new ChainCodeBucket(this.leveldb,ccid)
this.ccbucket += ccid -> tmp1
tmp1.Put(key, value)
computeMerkel
}
}
}
}
}
private def computeMerkel={
synchronized{
val start = System.currentTimeMillis()
var value : Array[Byte] = null
this.ccids.foreach(f=>{
var tmp = this.ccbucket(f)
if(tmp != null){
val tv = tmp.getCCMerkle
if(tv != null){
if(value == null){
value = tv
}else{
value = Array.concat(value , tv)
}
}
}
})
if(value != null) this.MerkleHash = Sha256.hash(value)
val end = System.currentTimeMillis()
log.info("Merkle compute time="+(end - start)+"ms")
}
}
def getMerkleHash:Array[Byte]={
this.MerkleHash
}
def getMerkelHash4String:String={
var rel:String = ""
val bb = getMerkleHash
if(bb != null){
rel = BytesHex.bytes2hex(bb)
}
rel
}
def save={
val keyhash = IdxPrefix.WorldStateForInternetPrefix + "merkletreehash"
this.leveldb.Put(keyhash, this.MerkleHash)
val key = IdxPrefix.WorldStateForInternetPrefix + IdxPrefix.ChainCodeList
val cckb = SerializeUtils.serialise(this.ccids)
this.leveldb.Put(key, cckb)
this.ccids.foreach(f=>{
var tmp = this.ccbucket(f)
if(tmp != null){
tmp.save
}
})
}
private def getMerkleHashFromDB={
val key = IdxPrefix.WorldStateForInternetPrefix + "merkletreehash"
val v = leveldb.Get(key)
if(v != null){
this.MerkleHash = v
}
}
private def getAllChainCode={
val key = IdxPrefix.WorldStateForInternetPrefix + IdxPrefix.ChainCodeList
val v = leveldb.Get(key)
if(v != null){
val tmp = SerializeUtils.deserialise(v)
if(tmp.isInstanceOf[mutable.ArrayBuffer[String]] ){
var ta = tmp.asInstanceOf[mutable.ArrayBuffer[String]]
this.ccids = ta
}
}
}
}

1002
src/main/scala/rep/storage/test/test.scala
View File

File diff suppressed because it is too large Load Diff