Merge "Bug 1306: Fixed incorrect AD/MD-SAL flow conversion."
authorEd Warnicke <eaw@cisco.com>
Fri, 4 Jul 2014 20:44:19 +0000 (20:44 +0000)
committerGerrit Code Review <gerrit@opendaylight.org>
Fri, 4 Jul 2014 20:44:19 +0000 (20:44 +0000)
29 files changed:
opendaylight/commons/opendaylight/pom.xml
opendaylight/distribution/opendaylight/src/main/resources/functions.sh [new file with mode: 0644]
opendaylight/distribution/opendaylight/src/main/resources/run.bat
opendaylight/distribution/opendaylight/src/main/resources/run.sh
opendaylight/md-sal/sal-distributed-datastore/pom.xml
opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/Shard.java
opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardTransaction.java
opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardTransactionChain.java
opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ThreePhaseCommitCohort.java
opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ThreePhaseCommitCohortProxy.java
opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/TransactionChainProxy.java
opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/TransactionProxy.java
opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/exceptions/PrimaryNotFoundException.java [new file with mode: 0644]
opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/exceptions/TimeoutException.java [new file with mode: 0644]
opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/utils/ActorContext.java
opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/config/yang/config/distributed_datastore_provider/DistributedConfigDataStoreProviderModule.java [new file with mode: 0644]
opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/config/yang/config/distributed_datastore_provider/DistributedConfigDataStoreProviderModuleFactory.java [moved from opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/config/yang/config/distributed_datastore_provider/DistributedDataStoreProviderModuleFactory.java with 52% similarity]
opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/config/yang/config/distributed_datastore_provider/DistributedDataStoreProviderModule.java [deleted file]
opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/config/yang/config/distributed_datastore_provider/DistributedOperationalDataStoreProviderModule.java [new file with mode: 0644]
opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/config/yang/config/distributed_datastore_provider/DistributedOperationalDataStoreProviderModuleFactory.java [new file with mode: 0644]
opendaylight/md-sal/sal-distributed-datastore/src/main/resources/application.conf [new file with mode: 0644]
opendaylight/md-sal/sal-distributed-datastore/src/main/yang/distributed-datastore-provider.yang
opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/BasicIntegrationTest.java
opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/DistributedDataStoreIntegrationTest.java [new file with mode: 0644]
opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/ShardTest.java
opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/ShardTransactionTest.java
opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/ThreePhaseCommitCohortProxyTest.java [new file with mode: 0644]
opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/utils/TestUtils.java [new file with mode: 0644]
opendaylight/md-sal/statistics-manager/src/main/java/org/opendaylight/controller/md/statistics/manager/StatisticsRequestScheduler.java

index de8caa0..3aeb2cc 100644 (file)
@@ -43,7 +43,6 @@
     <commons.catalina.ha>7.0.32.v201211201952</commons.catalina.ha>
     <commons.catalina.tribes>7.0.32.v201211201952</commons.catalina.tribes>
     <commons.checkstyle.version>0.0.3-SNAPSHOT</commons.checkstyle.version>
-    <commons.codec.version>1.7</commons.codec.version>
     <commons.coyote>7.0.32.v201211201952</commons.coyote>
     <commons.el>7.0.32.v201211081135</commons.el>
     <commons.fileupload.version>1.2.2</commons.fileupload.version>
diff --git a/opendaylight/distribution/opendaylight/src/main/resources/functions.sh b/opendaylight/distribution/opendaylight/src/main/resources/functions.sh
new file mode 100644 (file)
index 0000000..21dd4c1
--- /dev/null
@@ -0,0 +1,18 @@
+#!/bin/bash
+
+# Function harvestHelp searches in run.sh part for line starting with "##<name command>".
+# Next lines starting with "#<text>" will be printed without first char # (=help content).
+# Help content has to end with "##" on new line.
+# Example:
+##foo
+#   Foo is world wide used synnonym for bar.
+##
+function harvestHelp() {
+    key="$1"
+    if [ -z "${key}" ]; then
+         key='HELP'
+    fi
+    echo
+    sed -rn "/^##${key}$/,/^##/ p" $0 | sed -r '1 d; $ d; s/^#/  /'
+    grep "##${key}" $0 > /dev/null
+}
index 72ccd02..ce13e33 100644 (file)
@@ -14,6 +14,7 @@ SET jvmMaxMemory=
 SET extraJVMOpts=
 SET consoleOpts=-console -consoleLog
 SET PID=
+SET JAVA_H=%JAVA_HOME%\bin\jps.exe
 
 :LOOP
 IF "%~1" NEQ "" (
@@ -61,7 +62,7 @@ IF "%~1" NEQ "" (
        GOTO :LOOP
     )
     IF "!CARG!"=="-status" (
-       for /F "TOKENS=1" %%G in ('%JAVA_HOME%\bin\jps.exe -lvV ^| find /I "opendaylight"') do (
+       for /F "TOKENS=1" %%G in ('""!JAVA_H!" -lvV ^| find /I "opendaylight""') do (
            set PID=%%G
        )
        if "!PID!" NEQ "" (
@@ -72,7 +73,7 @@ IF "%~1" NEQ "" (
        GOTO :EOF
     )
     IF "!CARG!"=="-stop" (
-       for /F "TOKENS=1" %%G in ('%JAVA_HOME%\bin\jps.exe -lvV ^| find /I "opendaylight"') do (
+       for /F "TOKENS=1" %%G in ('""!JAVA_H!" -lvV ^| find /I "opendaylight""') do (
            set PID=%%G
        )
        if "!PID!" NEQ "" (
@@ -89,18 +90,23 @@ IF "%~1" NEQ "" (
        GOTO :LOOP
     )
     IF "!CARG:~0,2!"=="-D" (
-       SET extraJVMOpts=%extraJVMOpts% !CARG!
+       SET extraJVMOpts=!extraJVMOpts! !CARG!
        SHIFT
        GOTO :LOOP
     )
     IF "!CARG:~0,2!"=="-X" (
-       SET extraJVMOpts=%extraJVMOpts% !CARG!
+       SET extraJVMOpts=!extraJVMOpts! !CARG!
        SHIFT
        GOTO :LOOP
     )
     IF "!CARG!"=="-help" (
-        ECHO "Usage: %0 [-jmx] [-jmxport <num>] [-debug] [-debugsuspend] [-debugport <num>] [-start] [-consoleport <num>]] [-stop] [-status] [-console] [-help] [<other args will automatically be used for the JVM>]"
-        ECHO Note: Enclose any JVM or System properties within double quotes.
+        SHIFT
+        SET CARG=%2
+        IF "!CARG!" NEQ "" (
+             CALL:!CARG!
+        ) ELSE (
+              CALL:helper
+         )
         GOTO :EOF
     )
 
@@ -110,28 +116,24 @@ IF "%~1" NEQ "" (
 
 IF "%debugEnabled%" NEQ "" (
     REM ECHO "DEBUG enabled"
-    SET extraJVMOpts=%extraJVMOpts% -Xdebug -Xrunjdwp:transport=dt_socket,server=y,suspend=n,address=%debugport%
+    SET extraJVMOpts=!extraJVMOpts! -Xdebug -Xrunjdwp:transport=dt_socket,server=y,suspend=n,address=%debugport%
 )
+
 IF "%debugSuspended%" NEQ "" (
     REM ECHO "DEBUG enabled suspended"
-    SET extraJVMOpts=%extraJVMOpts% -Xdebug -Xrunjdwp:transport=dt_socket,server=y,suspend=y,address=%debugport%
+    SET extraJVMOpts=!extraJVMOpts! -Xdebug -Xrunjdwp:transport=dt_socket,server=y,suspend=y,address=%debugport%
 )
 
 IF "%jvmMaxMemory%"=="" (
     SET jvmMaxMemory=-Xmx1G
-    ECHO *****************************************************************
-    ECHO JVM maximum memory was not defined. Setting maximum memory to 1G.
-    ECHO To define the maximum memory, specify the -Xmx setting on the
-    ECHO command line.
-    ECHO                    e.g. run.bat -Xmx1G
-    ECHO *****************************************************************"
+    ECHO Setting maximum memory to 1G.
 )
 
-SET extraJVMOpts=%extraJVMOpts%  %jvmMaxMemory%
+SET extraJVMOpts=!extraJVMOpts!  %jvmMaxMemory%
 
 IF "%jmxEnabled%" NEQ "" (
     REM ECHO "JMX enabled "
-    SET extraJVMOpts=%extraJVMOpts% -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Dcom.sun.management.jmxremote.port=%jmxport% -Dcom.sun.management.jmxremote
+    SET extraJVMOpts=!extraJVMOpts! -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Dcom.sun.management.jmxremote.port=%jmxport% -Dcom.sun.management.jmxremote
 )
 IF "%startEnabled%" NEQ "" (
     REM ECHO "START enabled "
@@ -139,7 +141,7 @@ IF "%startEnabled%" NEQ "" (
 )
 
 REM       Check if controller is already running
-for /F "TOKENS=1" %%G in ('%JAVA_HOME%\bin\jps.exe -lvV ^| find /I "opendaylight"') do (
+for /F "TOKENS=1" %%G in ('""!JAVA_H!" -lvV ^| find /I "opendaylight""') do (
     SET PID=%%G
 )
 if "!PID!" NEQ "" (
@@ -154,16 +156,125 @@ SET cp="%basedir%lib\org.eclipse.osgi-3.8.1.v20120830-144521.jar;%basedir%lib\or
 REM       Now set framework classpath
 SET fwcp="file:\%basedir%lib\org.eclipse.osgi-3.8.1.v20120830-144521.jar,file:\%basedir%lib\org.eclipse.virgo.kernel.equinox.extensions-3.6.0.RELEASE.jar,file:\%basedir%lib\org.eclipse.equinox.launcher-1.3.0.v20120522-1813.jar"
 
-SET RUN_CMD="%JAVA_HOME%\bin\java.exe" -Dopendaylight.controller %extraJVMOpts% -Djava.io.tmpdir="%basedir%work\tmp" -Djava.awt.headless=true -Dosgi.install.area=%basedir% -Dosgi.configuration.area="%basedir%configuration" -Dosgi.frameworkClassPath=%fwcp% -Dosgi.framework="file:\%basedir%lib\org.eclipse.osgi-3.8.1.v20120830-144521.jar" -classpath %cp% org.eclipse.equinox.launcher.Main %consoleOpts%
+SET RUN_CMD="%JAVA_HOME%\bin\java.exe" -Dopendaylight.controller !extraJVMOpts! -Djava.io.tmpdir="%basedir%work\tmp" -Djava.awt.headless=true -Dosgi.install.area=%basedir% -Dosgi.configuration.area="%basedir%configuration" -Dosgi.frameworkClassPath=%fwcp% -Dosgi.framework="file:\%basedir%lib\org.eclipse.osgi-3.8.1.v20120830-144521.jar" -classpath %cp% org.eclipse.equinox.launcher.Main %consoleOpts%
 
-ECHO %RUN_CMD%
+ECHO !RUN_CMD!
 
 if "%startEnabled%" NEQ "" (
-    START /B cmd /C CALL %RUN_CMD% > %basedir%\logs\controller.out 2>&1
+    START /B cmd /C CALL !RUN_CMD! > %basedir%\logs\controller.out 2>&1
     ECHO Running controller in the background.
+    EXIT /B 1
 ) else (
-    %RUN_CMD%
+    !RUN_CMD!
     EXIT /B %ERRORLEVEL%
 )
 
+:helper
+echo. For more information on a specific command, type -help command-name.
+echo.
+echo   jmx              ^[-jmx^]
+echo   jmxport          ^[-jmxport ^<num^>^] - DEFAULT is 1088
+echo   debug            ^[-debug^]
+echo   debugsuspend     ^[-debugsuspend^]
+echo   debugport        ^[-debugport ^<num^>^] - DEFAULT is 8000
+echo   start            ^[-start ^[^<console port^>^]^] - DEFAULT port is 2400
+echo   stop             ^[-stop^]
+echo   status           ^[-status^]
+echo   console          ^[-console^]
+echo   agentpath        ^[-agentpath:^<path to lib^>^]
+exit/B 1
+
+:debugsuspend
+ECHO.
+ECHO. debugsuspend     ^[-debugsuspend^]
+ECHO.
+ECHO. This command sets suspend on true in runjdwp in extra JVM options. If its true, VMStartEvent has a suspendPolicy of SUSPEND_ALL. If its false, VMStartEvent has a suspendPolicy of SUSPEND_NONE.
+ECHO.
+EXIT /B 1
+
+:debugport
+ECHO.
+ECHO. debugport        ^[-debugport ^<num^>^] - DEFAULT is 8000
+ECHO.
+ECHO. Set address for settings in runjdwp in extra JVM options.
+ECHO. The address is transport address for the connection.
+ECHO. The address has to be in the range ^[1024,65535^]. If the option was not call, port will be set to default value.
+ECHO.
+EXIT /B 1
+
+:jmxport
+ECHO.
+ECHO. jmxport          ^[-jmxport ^<num^>^] - DEFAULT is 1088
+ECHO.
+ECHO.    Set jmx port for com.sun.management.jmxremote.port in JMX support. Port has to be in the range ^[1024,65535^]. If this option was not call, port will be set to default value.
+ECHO.
+EXIT /B 1
+
+:debug
+ECHO.
+ECHO. debug            [-debug]
+ECHO.
+ECHO. Run ODL controller with -Xdebug and -Xrunjdwp:transport=dt_socket,server=y,suspend=n,address=^$^{debugport^}
+ECHO.
+ECHO.    -Xdebug enables debugging capabilities in the JVM which are used by the Java Virtual Machine Tools Interface (JVMTI). JVMTI is a low-level debugging interface used by debuggers and profiling tools.
+ECHO.
+ECHO.    -Xrunjdwp option loads the JPDA reference implementation of JDWP. This library resides in the target VM and uses JVMDI and JNI to interact with it. It uses a transport and the JDWP protocol to communicate with a separate debugger application.
+ECHO.
+ECHO. settings for -Xrunjdwp:
+ECHO.            transport -  name of the transport to use in connecting to debugger application
+ECHO.            server    -  if 'y', listen for a debugger application to attach; otherwise, attach to the debugger application at the specified address
+ECHO.                      -  if 'y' and no address is specified, choose a transport address at which to listen for a debugger application, and print the address to the standard output stream
+ECHO.            suspend   -  if 'y', VMStartEvent has a suspend Policy of SUSPEND_ALL
+ECHO.                      -  if 'n', VMStartEvent has a suspend policy of SUSPEND_NONE
+ECHO.            address   -  transport address for the connection
+ECHO.                      -  if server=n, attempt to attach to debugger application at this address
+ECHO.          -  if server=y, listen for a connection at this address
+ECHO.
+EXIT /B 1
+
+:jmx
+ECHO.
+ECHO. jmx              [-jmx]
+ECHO.
+ECHO. Add JMX support. With settings for extra JVM options: -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Dcom.sun.management.jmxremote.port=^$^{jmxport^} -Dcom.sun.management.jmxremote
+ECHO. jmxport can by set with option -jmxport ^<num^>. Default num for the option is 1088.
+ECHO.
+EXIT /B 1
+
+:stop
+ECHO.
+ECHO. stop             ^[-stop^]
+ECHO.
+ECHO. If a controller is running, the command stop controller. Pid will be clean.
+ECHO.
+EXIT /B 1
+
+:status
+ECHO.
+ECHO. status           ^[-status^]
+ECHO.
+ECHO. Find out whether a controller is running and print it.
+ECHO.
+EXIT /B 1
+
+:start
+ECHO.
+ECHO. start            ^[-start ^[^<console port^>^]^]
+ECHO.
+ECHO.    If controller is not running, the command with argument^(for set port, where controller has start^) will start new controller on a port. The port has to be in the range ^[1024,65535^]. If this option was not call, port will be set to default value. Pid will be create.
+EXIT /B 1
+
+:console
+ECHO.
+ECHO. console          [-console]
+ECHO.     Default option.
+EXIT /B 1
+
+:agentpath
+ECHO.
+ECHO. agentpath        ^[-agentpath:^<path to lib^>^]
+ECHO.
+ECHO.    Agentpath option passes path to agent to jvm in order to load native agent library, e.g. yourkit profiler agent.
+EXIT /B 1
+
 
index 1e903d0..13be233 100755 (executable)
@@ -1,5 +1,20 @@
 #!/bin/bash
 
+##HELP
+# For more information on a specific command, type -help command-name.
+#
+#   jmx              [-jmx]
+#   jmxport          [-jmxport <num>] - DEFAULT is 1088
+#   debug            [-debug]
+#   debugsuspend     [-debugsuspend]
+#   debugport        [-debugport <num>] - DEFAULT is 8000
+#   start            [-start [<console port>]] - DEFAULT port is 2400
+#   stop             [-stop]
+#   status           [-status]
+#   console          [-console]
+#   agentpath        [-agentpath:<path to lib>]
+##
+
 platform='unknown'
 unamestr=`uname`
 if [[ "$unamestr" == 'Linux' ]]; then
@@ -58,11 +73,6 @@ else
     datadir=${ODL_DATADIR}
 fi
 
-function usage {
-    echo "Usage: $0 [-jmx] [-jmxport <num>] [-debug] [-debugsuspend] [-debugport <num>] [-start [<console port>]] [-stop] [-status] [-console] [-help] [-agentpath:<path to lib>] [<other args will automatically be used for the JVM>]"
-    exit 1
-}
-
 if [ -z ${TMP} ]; then
     pidfile="/tmp/opendaylight.PID"
 else
@@ -82,13 +92,15 @@ stopdaemon=0
 statusdaemon=0
 consolestart=1
 dohelp=0
-jvmMaxMemory=""
+jvmMaxMemory="-Xmx1G"
 extraJVMOpts=""
 agentPath=""
 unknown_option=0
+helper=""
 while true ; do
     case "$1" in
         -debug) debug=1; shift ;;
+        -help) dohelp=1; shift;  helper=$1; break ;;
         -jmx) startjmx=1; shift ;;
         -debugsuspend) debugsuspend=1; shift ;;
         -debugport) shift; debugportread="$1"; if [[ "${debugportread}" =~ ^[0-9]+$ ]] ; then debugport=${debugportread}; shift; else echo "-debugport expects a number but was not found"; exit -1; fi;;
@@ -97,37 +109,39 @@ while true ; do
         -stop) stopdaemon=1; shift ;;
         -status) statusdaemon=1; shift ;;
         -console) shift ;;
-        -help) dohelp=1; shift;;
         -Xmx*) jvmMaxMemory="$1"; shift;;
         -D*) extraJVMOpts="${extraJVMOpts} $1"; shift;;
         -X*) extraJVMOpts="${extraJVMOpts} $1"; shift;;
         -agentpath:*) agentPath="$1"; shift;;
         "") break ;;
-        *) echo "Unknown option $1"; unknown_option=1; shift ;;
+        *) echo "Unknown option $1"; unknown_option=1; break ;;
     esac
 done
 
-# Unknown Options and help
+
+
 if [ "${unknown_option}" -eq 1 ]; then
-    usage
+    echo "Use -help for more information."
+    exit 1
 fi
 
-if [ "${dohelp}" -eq 1 ]; then
-    usage
-fi
 
-if [ "${jvmMaxMemory}" == "" ]; then
-    jvmMaxMemory="-Xmx1G"
-    echo "*****************************************************************"
-    echo "JVM maximum memory was not defined. Setting maximum memory to 1G."
-    echo "To define the maximum memory, specify the -Xmx setting on the"
-    echo "command line. "
-    echo "        e.g. ./run.sh -Xmx1G"
-    echo "*****************************************************************"
+if [ "${dohelp}" -eq 1 ]; then
+    . ${basedir}/functions.sh
+    harvestHelp ${helper}
+    echo -e '\nFor other information type -help.\n'
+    exit 1
 fi
 
 extraJVMOpts="${extraJVMOpts} ${jvmMaxMemory}"
 
+##debugport
+#debugport        [-debugport <num>] - DEFAULT is 8000
+#
+#    Set address for settings in runjdwp in extra JVM options.
+#    The address is transport address for the connection.
+#    The address has to be in the range [1024,65535]. If this option was not call, port will be set to default value.
+##
 # Validate debug port
 if [[ "${debugport}" -lt 1024 ]] || [[ "${debugport}" -gt 65535 ]]; then
     echo "Debug Port not in the range [1024,65535] ${debugport}"
@@ -140,19 +154,51 @@ if [[ "${daemonport}" -lt 1024 ]] || [[ "${daemonport}" -gt 65535 ]]; then
     exit -1
 fi
 
+##jmxport
+#jmxport          [-jmxport <num>] - DEFAULT is 1088
+#
+#    Set jmx port for com.sun.management.jmxremote.port in JMX support. Port has to be in the range [1024,65535]. If this option was not call, port will be set to default value.
+##
 # Validate jmx port
 if [[ "${jmxport}" -lt 1024 ]] || [[ "${jmxport}" -gt 65535 ]]; then
     echo "JMX Port not in the range [1024,65535] value is ${jmxport}"
     exit -1
 fi
+##debug
+#debug            [-debug]
+#
+#Run ODL controller with -Xdebug and -Xrunjdwp:transport=dt_socket,server=y,suspend=n,address=\${debugport}
+#-Xdebug enables debugging capabilities in the JVM which are used by the Java Virtual Machine Tools Interface (JVMTI). JVMTI is a low-level debugging interface used by debuggers and profiling tools.
+#-Xrunjdwp option loads the JPDA reference implementation of JDWP. This library resides in the target VM and uses JVMDI and JNI to interact with it. It uses a transport and the JDWP protocol to
+#communicate with a separate debugger application.
+#settings for -Xrunjdwp:
+#            transport -  name of the transport to use in connecting to debugger application
+#            server    -  if “y”, listen for a debugger application to attach; otherwise, attach to the debugger application at the specified address
+#                      -  if “y” and no address is specified, choose a transport address at which to listen for a debugger application, and print the address to the standard output stream
+#            suspend   -  if “y”, VMStartEvent has a suspend Policy of SUSPEND_ALL
+#                      -  if “n”, VMStartEvent has a suspend policy of SUSPEND_NONE
+#            address   -  transport address for the connection
+#                      -  if server=n, attempt to attach to debugger application at this address
+#          -  if server=y, listen for a connection at this address
+##
 
+##debugsuspend
+#debugsuspend     [-debugsuspend]
+#
+#This command sets suspend on true in runjdwp in extra JVM options. If its true, VMStartEvent has a suspendPolicy of SUSPEND_ALL. If its false, VMStartEvent has a suspendPolicy of SUSPEND_NONE.
+##
 # Debug options
 if [ "${debugsuspend}" -eq 1 ]; then
     extraJVMOpts="${extraJVMOpts} -Xdebug -Xrunjdwp:transport=dt_socket,server=y,suspend=y,address=${debugport}"
 elif [ "${debug}" -eq 1 ]; then
     extraJVMOpts="${extraJVMOpts} -Xdebug -Xrunjdwp:transport=dt_socket,server=y,suspend=n,address=${debugport}"
 fi
-
+##jmx
+#jmx              [-jmx]
+#
+#Add JMX support. With settings for extra JVM options: -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Dcom.sun.management.jmxremote.port=\${jmxport} -Dcom.sun.management.jmxremote
+#jmxport can by set with command -jmxport <num>. Default num for the option is 1088.
+##
 # Add JMX support
 if [ "${startjmx}" -eq 1 ]; then
     extraJVMOpts="${extraJVMOpts} -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Dcom.sun.management.jmxremote.port=${jmxport} -Dcom.sun.management.jmxremote"
@@ -182,6 +228,11 @@ FWCLASSPATH=${FWCLASSPATH},file:${basedir}/lib/org.eclipse.equinox.launcher-1.3.
 
 cd $basedir
 
+##stop
+#stop             [-stop]
+#
+#If a controller is running, the command stop controller. Pid will be clean.
+##
 if [ "${stopdaemon}" -eq 1 ]; then
     if [ -e "${pidfile}" ]; then
         daemonpid=`cat "${pidfile}"`
@@ -195,6 +246,11 @@ if [ "${stopdaemon}" -eq 1 ]; then
     fi
 fi
 
+##status
+#status           [-status]
+#
+#Find out whether a controller is running and print it.
+##
 if [ "${statusdaemon}" -eq 1 ]; then
     if [ -e "${pidfile}" ]; then
         daemonpid=`cat "${pidfile}"`
@@ -219,6 +275,22 @@ bdir=`echo "${basedir}" | sed 's/ /\\ /g'`
 confarea=`echo "${datadir}" | sed 's/ /\\ /g'`
 fwclasspath=`echo "${FWCLASSPATH}" | sed 's/ /\\ /g'`
 
+##start
+#start            [-start [<console port>]]
+#
+#    If controller is not running, the command with argument(for set port, where controller has start) will start new controller on a port. The port has to be in the range [1024,65535]. If this option was not call, port will be set to default value. Pid will be create.
+##
+##console
+#console          [-console]
+#
+#    Default option.
+##
+##agentpath
+#agentpath        [-agentpath:<path to lib>]
+#
+#   Agentpath option passes path to agent to jvm in order to load native agent library, e.g. yourkit profiler agent.
+##
+echo "JVM maximum memory was set to ${jvmMaxMemory}."
 if [ "${startdaemon}" -eq 1 ]; then
     if [ -e "${pidfile}" ]; then
         echo "Another instance of controller running, check with $0 -status"
index ea686d9..5aea3fb 100644 (file)
       <artifactId>sal-binding-config</artifactId>
     </dependency>
 
-    <!--
-      Adding a temporary dependency on the sal-broker-impl so that we can use InMemoryDOMDataStore
+      <dependency>
+          <groupId>org.opendaylight.yangtools</groupId>
+          <artifactId>yang-data-api</artifactId>
+      </dependency>
+
+      <dependency>
+      <groupId>org.opendaylight.yangtools</groupId>
+      <artifactId>yang-data-impl</artifactId>
+      </dependency>
 
-      InMemoryDOMDataStore needs to be moved into its own module and be wired up using config subsystem before
-      this bundle can use it
-    -->
     <dependency>
       <groupId>org.opendaylight.controller</groupId>
-      <artifactId>sal-broker-impl</artifactId>
+      <artifactId>sal-inmemory-datastore</artifactId>
+      <version>1.1-SNAPSHOT</version>
     </dependency>
 
     <dependency>
index b4ad089..09ad005 100644 (file)
@@ -20,8 +20,10 @@ import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.ListeningExecutorService;
 import com.google.common.util.concurrent.MoreExecutors;
 import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply;
+import org.opendaylight.controller.cluster.datastore.messages.CreateTransaction;
 import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionChain;
 import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionChainReply;
+import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionReply;
 import org.opendaylight.controller.cluster.datastore.messages.ForwardedCommitTransaction;
 import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListener;
 import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListenerReply;
@@ -29,6 +31,7 @@ import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContex
 import org.opendaylight.controller.cluster.datastore.modification.Modification;
 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataChangeListener;
 import org.opendaylight.controller.md.sal.dom.store.impl.InMemoryDOMDataStore;
+import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction;
 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
 import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionChain;
 import org.opendaylight.yangtools.yang.data.api.InstanceIdentifier;
@@ -61,6 +64,8 @@ public class Shard extends UntypedProcessor {
         Logging.getLogger(getContext().system(), this);
 
     private Shard(String name) {
+        log.info("Creating shard : {}", name );
+
         store = new InMemoryDOMDataStore(name, storeExecutor);
     }
 
@@ -77,6 +82,8 @@ public class Shard extends UntypedProcessor {
 
     @Override
     public void onReceive(Object message) throws Exception {
+        log.debug("Received message {}", message);
+
         if (message instanceof CreateTransactionChain) {
             createTransactionChain();
         } else if (message instanceof RegisterChangeListener) {
@@ -87,9 +94,21 @@ public class Shard extends UntypedProcessor {
             handleForwardedCommit((ForwardedCommitTransaction) message);
         } else if (message instanceof Persistent) {
             commit((Persistent) message);
+        } else if (message instanceof CreateTransaction) {
+            createTransaction();
         }
     }
 
+    private void createTransaction() {
+        DOMStoreReadWriteTransaction transaction =
+            store.newReadWriteTransaction();
+        ActorRef transactionActor = getContext().actorOf(
+            ShardTransaction.props(transaction, getSelf()));
+        getSender()
+            .tell(new CreateTransactionReply(transactionActor.path()),
+                getSelf());
+    }
+
     private void commit(Persistent message) {
         Modification modification = (Modification) message.payload();
         DOMStoreThreePhaseCommitCohort cohort =
index 75744ca..a2da063 100644 (file)
@@ -9,6 +9,7 @@
 package org.opendaylight.controller.cluster.datastore;
 
 import akka.actor.ActorRef;
+import akka.actor.PoisonPill;
 import akka.actor.Props;
 import akka.actor.UntypedActor;
 import akka.event.Logging;
@@ -36,6 +37,7 @@ import org.opendaylight.controller.cluster.datastore.modification.MutableComposi
 import org.opendaylight.controller.cluster.datastore.modification.WriteModification;
 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction;
 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
+import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionChain;
 import org.opendaylight.yangtools.yang.data.api.InstanceIdentifier;
 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
 
@@ -43,16 +45,16 @@ import java.util.concurrent.ExecutionException;
 
 /**
  * The ShardTransaction Actor represents a remote transaction
- *<p>
+ * <p>
  * The ShardTransaction Actor delegates all actions to DOMDataReadWriteTransaction
- *</p>
- *<p>
+ * </p>
+ * <p>
  * Even though the DOMStore and the DOMStoreTransactionChain implement multiple types of transactions
  * the ShardTransaction Actor only works with read-write transactions. This is just to keep the logic simple. At this
  * time there are no known advantages for creating a read-only or write-only transaction which may change over time
  * at which point we can optimize things in the distributed store as well.
- *</p>
- *<p>
+ * </p>
+ * <p>
  * Handles Messages <br/>
  * ---------------- <br/>
  * <li> {@link org.opendaylight.controller.cluster.datastore.messages.ReadData}
@@ -65,123 +67,164 @@ import java.util.concurrent.ExecutionException;
  */
 public class ShardTransaction extends UntypedActor {
 
-  private final ActorRef shardActor;
-
-  private final DOMStoreReadWriteTransaction transaction;
-
-  private final MutableCompositeModification modification = new MutableCompositeModification();
-
-  private final LoggingAdapter log = Logging.getLogger(getContext().system(), this);
-
-  public ShardTransaction(DOMStoreReadWriteTransaction transaction, ActorRef shardActor) {
-    this.transaction = transaction;
-    this.shardActor = shardActor;
-  }
-
-
-  public static Props props(final DOMStoreReadWriteTransaction transaction, final ActorRef shardActor){
-    return Props.create(new Creator<ShardTransaction>(){
-
-      @Override
-      public ShardTransaction create() throws Exception {
-        return new ShardTransaction(transaction, shardActor);
-      }
-    });
-  }
-
-  @Override
-  public void onReceive(Object message) throws Exception {
-    if(message instanceof ReadData){
-      readData((ReadData) message);
-    } else if(message instanceof WriteData){
-      writeData((WriteData) message);
-    } else if(message instanceof MergeData){
-      mergeData((MergeData) message);
-    } else if(message instanceof DeleteData){
-      deleteData((DeleteData) message);
-    } else if(message instanceof ReadyTransaction){
-      readyTransaction((ReadyTransaction) message);
-    } else if(message instanceof CloseTransaction){
-      closeTransaction((CloseTransaction) message);
-    } else if(message instanceof GetCompositedModification){
-      // This is here for testing only
-      getSender().tell(new GetCompositeModificationReply(new ImmutableCompositeModification(modification)), getSelf());
+    private final ActorRef shardActor;
+
+    // FIXME : see below
+    // If transactionChain is not null then this transaction is part of a
+    // transactionChain. Not really clear as to what that buys us
+    private final DOMStoreTransactionChain transactionChain;
+
+    private final DOMStoreReadWriteTransaction transaction;
+
+    private final MutableCompositeModification modification =
+        new MutableCompositeModification();
+
+    private final LoggingAdapter log =
+        Logging.getLogger(getContext().system(), this);
+
+    public ShardTransaction(DOMStoreReadWriteTransaction transaction,
+        ActorRef shardActor) {
+        this(null, transaction, shardActor);
+    }
+
+    public ShardTransaction(DOMStoreTransactionChain transactionChain, DOMStoreReadWriteTransaction transaction,
+        ActorRef shardActor) {
+        this.transactionChain = transactionChain;
+        this.transaction = transaction;
+        this.shardActor = shardActor;
+    }
+
+
+
+    public static Props props(final DOMStoreReadWriteTransaction transaction,
+        final ActorRef shardActor) {
+        return Props.create(new Creator<ShardTransaction>() {
+
+            @Override
+            public ShardTransaction create() throws Exception {
+                return new ShardTransaction(transaction, shardActor);
+            }
+        });
+    }
+
+    public static Props props(final DOMStoreTransactionChain transactionChain, final DOMStoreReadWriteTransaction transaction,
+        final ActorRef shardActor) {
+        return Props.create(new Creator<ShardTransaction>() {
+
+            @Override
+            public ShardTransaction create() throws Exception {
+                return new ShardTransaction(transactionChain, transaction, shardActor);
+            }
+        });
     }
-  }
-
-  private void readData(ReadData message) {
-    final ActorRef sender = getSender();
-    final ActorRef self = getSelf();
-    final InstanceIdentifier path = message.getPath();
-    final ListenableFuture<Optional<NormalizedNode<?, ?>>> future = transaction.read(path);
-
-    future.addListener(new Runnable() {
-      @Override
-      public void run() {
-        try {
-          Optional<NormalizedNode<?, ?>> optional = future.get();
-          if(optional.isPresent()){
-            sender.tell(new ReadDataReply(optional.get()), self);
-          } else {
-            //TODO : Need to decide what to do here
-          }
-        } catch (InterruptedException | ExecutionException e) {
-          log.error(e, "An exception happened when reading data from path : " + path.toString());
-        }
 
-      }
-    }, getContext().dispatcher());
-  }
 
+    @Override
+    public void onReceive(Object message) throws Exception {
+        log.debug("Received message {}", message);
+
+        if (message instanceof ReadData) {
+            readData((ReadData) message);
+        } else if (message instanceof WriteData) {
+            writeData((WriteData) message);
+        } else if (message instanceof MergeData) {
+            mergeData((MergeData) message);
+        } else if (message instanceof DeleteData) {
+            deleteData((DeleteData) message);
+        } else if (message instanceof ReadyTransaction) {
+            readyTransaction((ReadyTransaction) message);
+        } else if (message instanceof CloseTransaction) {
+            closeTransaction((CloseTransaction) message);
+        } else if (message instanceof GetCompositedModification) {
+            // This is here for testing only
+            getSender().tell(new GetCompositeModificationReply(
+                new ImmutableCompositeModification(modification)), getSelf());
+        }
+    }
 
-  private void writeData(WriteData message){
-    modification.addModification(new WriteModification(message.getPath(), message.getData()));
-    transaction.write(message.getPath(), message.getData());
-    getSender().tell(new WriteDataReply(), getSelf());
-  }
+    private void readData(ReadData message) {
+        final ActorRef sender = getSender();
+        final ActorRef self = getSelf();
+        final InstanceIdentifier path = message.getPath();
+        final ListenableFuture<Optional<NormalizedNode<?, ?>>> future =
+            transaction.read(path);
+
+        future.addListener(new Runnable() {
+            @Override
+            public void run() {
+                try {
+                    Optional<NormalizedNode<?, ?>> optional = future.get();
+                    if (optional.isPresent()) {
+                        sender.tell(new ReadDataReply(optional.get()), self);
+                    } else {
+                        sender.tell(new ReadDataReply(null), self);
+                    }
+                } catch (InterruptedException | ExecutionException e) {
+                    log.error(e,
+                        "An exception happened when reading data from path : "
+                            + path.toString());
+                }
+
+            }
+        }, getContext().dispatcher());
+    }
 
-  private void mergeData(MergeData message){
-    modification.addModification(new MergeModification(message.getPath(), message.getData()));
-    transaction.merge(message.getPath(), message.getData());
-    getSender().tell(new MergeDataReply(), getSelf());
-  }
 
-  private void deleteData(DeleteData message){
-    modification.addModification(new DeleteModification(message.getPath()));
-    transaction.delete(message.getPath());
-    getSender().tell(new DeleteDataReply(), getSelf());
-  }
+    private void writeData(WriteData message) {
+        modification.addModification(
+            new WriteModification(message.getPath(), message.getData()));
+        transaction.write(message.getPath(), message.getData());
+        getSender().tell(new WriteDataReply(), getSelf());
+    }
 
-  private void readyTransaction(ReadyTransaction message){
-    DOMStoreThreePhaseCommitCohort cohort = transaction.ready();
-    ActorRef cohortActor = getContext().actorOf(ThreePhaseCommitCohort.props(cohort, shardActor, modification));
-    getSender().tell(new ReadyTransactionReply(cohortActor.path()), getSelf());
+    private void mergeData(MergeData message) {
+        modification.addModification(
+            new MergeModification(message.getPath(), message.getData()));
+        transaction.merge(message.getPath(), message.getData());
+        getSender().tell(new MergeDataReply(), getSelf());
+    }
 
-  }
+    private void deleteData(DeleteData message) {
+        modification.addModification(new DeleteModification(message.getPath()));
+        transaction.delete(message.getPath());
+        getSender().tell(new DeleteDataReply(), getSelf());
+    }
 
-  private void closeTransaction(CloseTransaction message){
-    transaction.close();
-    getSender().tell(new CloseTransactionReply(), getSelf());
-  }
+    private void readyTransaction(ReadyTransaction message) {
+        DOMStoreThreePhaseCommitCohort cohort = transaction.ready();
+        ActorRef cohortActor = getContext().actorOf(
+            ThreePhaseCommitCohort.props(cohort, shardActor, modification));
+        getSender()
+            .tell(new ReadyTransactionReply(cohortActor.path()), getSelf());
 
+    }
 
-  // These classes are in here for test purposes only
+    private void closeTransaction(CloseTransaction message) {
+        transaction.close();
+        getSender().tell(new CloseTransactionReply(), getSelf());
+        getSelf().tell(PoisonPill.getInstance(), getSelf());
+    }
 
-  static class GetCompositedModification {
 
-  }
+    // These classes are in here for test purposes only
 
-  static class GetCompositeModificationReply {
-    private final CompositeModification modification;
 
+    static class GetCompositedModification {
 
-    GetCompositeModificationReply(CompositeModification modification) {
-      this.modification = modification;
     }
 
 
-    public CompositeModification getModification() {
-      return modification;
+    static class GetCompositeModificationReply {
+        private final CompositeModification modification;
+
+
+        GetCompositeModificationReply(CompositeModification modification) {
+            this.modification = modification;
+        }
+
+
+        public CompositeModification getModification() {
+            return modification;
+        }
     }
-  }
 }
index 79aaa86..6c14f1d 100644 (file)
@@ -34,7 +34,7 @@ public class ShardTransactionChain extends UntypedActor{
   public void onReceive(Object message) throws Exception {
     if(message instanceof CreateTransaction){
       DOMStoreReadWriteTransaction transaction = chain.newReadWriteTransaction();
-      ActorRef transactionActor = getContext().actorOf(ShardTransaction.props(transaction, getContext().parent()));
+      ActorRef transactionActor = getContext().actorOf(ShardTransaction.props(chain, transaction, getContext().parent()));
       getSender().tell(new CreateTransactionReply(transactionActor.path()), getSelf());
     } else if (message instanceof CloseTransactionChain){
       chain.close();
index 61baf1a..e6adfbe 100644 (file)
@@ -9,6 +9,7 @@
 package org.opendaylight.controller.cluster.datastore;
 
 import akka.actor.ActorRef;
+import akka.actor.PoisonPill;
 import akka.actor.Props;
 import akka.actor.UntypedActor;
 import akka.event.Logging;
@@ -28,101 +29,111 @@ import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCoh
 
 import java.util.concurrent.ExecutionException;
 
-public class ThreePhaseCommitCohort extends UntypedActor{
-  private final DOMStoreThreePhaseCommitCohort cohort;
-  private final ActorRef shardActor;
-  private final CompositeModification modification;
-
-  public ThreePhaseCommitCohort(DOMStoreThreePhaseCommitCohort cohort, ActorRef shardActor, CompositeModification modification) {
-    this.cohort = cohort;
-    this.shardActor = shardActor;
-    this.modification = modification;
-  }
-
-  private final LoggingAdapter log = Logging.getLogger(getContext().system(), this);
-
-  public static Props props(final DOMStoreThreePhaseCommitCohort cohort, final ActorRef shardActor, final CompositeModification modification) {
-    return Props.create(new Creator<ThreePhaseCommitCohort>(){
-      @Override
-      public ThreePhaseCommitCohort create() throws Exception {
-        return new ThreePhaseCommitCohort(cohort, shardActor, modification);
-      }
-    });
-  }
-
-  @Override
-  public void onReceive(Object message) throws Exception {
-    if(message instanceof CanCommitTransaction){
-      canCommit((CanCommitTransaction) message);
-    } else if(message instanceof PreCommitTransaction) {
-      preCommit((PreCommitTransaction) message);
-    } else if(message instanceof CommitTransaction){
-      commit((CommitTransaction) message);
-    } else if (message instanceof AbortTransaction){
-      abort((AbortTransaction) message);
+public class ThreePhaseCommitCohort extends UntypedActor {
+    private final DOMStoreThreePhaseCommitCohort cohort;
+    private final ActorRef shardActor;
+    private final CompositeModification modification;
+
+    public ThreePhaseCommitCohort(DOMStoreThreePhaseCommitCohort cohort,
+        ActorRef shardActor, CompositeModification modification) {
+
+        this.cohort = cohort;
+        this.shardActor = shardActor;
+        this.modification = modification;
     }
-  }
-
-  private void abort(AbortTransaction message) {
-    final ListenableFuture<Void> future = cohort.abort();
-    final ActorRef sender = getSender();
-    final ActorRef self = getSelf();
-
-    future.addListener(new Runnable() {
-      @Override
-      public void run() {
-        try {
-          future.get();
-          sender.tell(new AbortTransactionReply(), self);
-        } catch (InterruptedException | ExecutionException e) {
-          log.error(e, "An exception happened when aborting");
-        }
-      }
-    }, getContext().dispatcher());
-  }
-
-  private void commit(CommitTransaction message) {
-    // Forward the commit to the shard
-    log.info("Commit transaction now + " + shardActor);
-    shardActor.forward(new ForwardedCommitTransaction(cohort, modification), getContext());
-
-  }
-
-  private void preCommit(PreCommitTransaction message) {
-    final ListenableFuture<Void> future = cohort.preCommit();
-    final ActorRef sender = getSender();
-    final ActorRef self = getSelf();
-
-    future.addListener(new Runnable() {
-      @Override
-      public void run() {
-        try {
-          future.get();
-          sender.tell(new PreCommitTransactionReply(), self);
-        } catch (InterruptedException | ExecutionException e) {
-          log.error(e, "An exception happened when preCommitting");
-        }
-      }
-    }, getContext().dispatcher());
-
-  }
-
-  private void canCommit(CanCommitTransaction message) {
-    final ListenableFuture<Boolean> future = cohort.canCommit();
-    final ActorRef sender = getSender();
-    final ActorRef self = getSelf();
-
-    future.addListener(new Runnable() {
-      @Override
-      public void run() {
-        try {
-          Boolean canCommit = future.get();
-          sender.tell(new CanCommitTransactionReply(canCommit), self);
-        } catch (InterruptedException | ExecutionException e) {
-          log.error(e, "An exception happened when aborting");
+
+    private final LoggingAdapter log =
+        Logging.getLogger(getContext().system(), this);
+
+    public static Props props(final DOMStoreThreePhaseCommitCohort cohort,
+        final ActorRef shardActor, final CompositeModification modification) {
+        return Props.create(new Creator<ThreePhaseCommitCohort>() {
+            @Override
+            public ThreePhaseCommitCohort create() throws Exception {
+                return new ThreePhaseCommitCohort(cohort, shardActor,
+                    modification);
+            }
+        });
+    }
+
+    @Override
+    public void onReceive(Object message) throws Exception {
+        log.debug("Received message {}", message);
+
+        if (message instanceof CanCommitTransaction) {
+            canCommit((CanCommitTransaction) message);
+        } else if (message instanceof PreCommitTransaction) {
+            preCommit((PreCommitTransaction) message);
+        } else if (message instanceof CommitTransaction) {
+            commit((CommitTransaction) message);
+        } else if (message instanceof AbortTransaction) {
+            abort((AbortTransaction) message);
         }
-      }
-    }, getContext().dispatcher());
+    }
+
+    private void abort(AbortTransaction message) {
+        final ListenableFuture<Void> future = cohort.abort();
+        final ActorRef sender = getSender();
+        final ActorRef self = getSelf();
+
+        future.addListener(new Runnable() {
+            @Override
+            public void run() {
+                try {
+                    future.get();
+                    sender.tell(new AbortTransactionReply(), self);
+                } catch (InterruptedException | ExecutionException e) {
+                    log.error(e, "An exception happened when aborting");
+                }
+            }
+        }, getContext().dispatcher());
+    }
 
-  }
+    private void commit(CommitTransaction message) {
+        // Forward the commit to the shard
+        log.debug("Forward commit transaction to Shard {} ", shardActor);
+        shardActor.forward(new ForwardedCommitTransaction(cohort, modification),
+            getContext());
+
+        getContext().parent().tell(PoisonPill.getInstance(), getSelf());
+
+    }
+
+    private void preCommit(PreCommitTransaction message) {
+        final ListenableFuture<Void> future = cohort.preCommit();
+        final ActorRef sender = getSender();
+        final ActorRef self = getSelf();
+
+        future.addListener(new Runnable() {
+            @Override
+            public void run() {
+                try {
+                    future.get();
+                    sender.tell(new PreCommitTransactionReply(), self);
+                } catch (InterruptedException | ExecutionException e) {
+                    log.error(e, "An exception happened when preCommitting");
+                }
+            }
+        }, getContext().dispatcher());
+
+    }
+
+    private void canCommit(CanCommitTransaction message) {
+        final ListenableFuture<Boolean> future = cohort.canCommit();
+        final ActorRef sender = getSender();
+        final ActorRef self = getSelf();
+
+        future.addListener(new Runnable() {
+            @Override
+            public void run() {
+                try {
+                    Boolean canCommit = future.get();
+                    sender.tell(new CanCommitTransactionReply(canCommit), self);
+                } catch (InterruptedException | ExecutionException e) {
+                    log.error(e, "An exception happened when aborting");
+                }
+            }
+        }, getContext().dispatcher());
+
+    }
 }
index 197b3b7..d12dc2b 100644 (file)
@@ -9,11 +9,28 @@
 package org.opendaylight.controller.cluster.datastore;
 
 import akka.actor.ActorPath;
+import akka.actor.ActorSelection;
 import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.ListenableFutureTask;
+import org.opendaylight.controller.cluster.datastore.exceptions.TimeoutException;
+import org.opendaylight.controller.cluster.datastore.messages.AbortTransaction;
+import org.opendaylight.controller.cluster.datastore.messages.AbortTransactionReply;
+import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransaction;
+import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransactionReply;
+import org.opendaylight.controller.cluster.datastore.messages.CommitTransaction;
+import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply;
+import org.opendaylight.controller.cluster.datastore.messages.PreCommitTransaction;
+import org.opendaylight.controller.cluster.datastore.messages.PreCommitTransactionReply;
+import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.Collections;
 import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 
 /**
  * ThreePhaseCommitCohortProxy represents a set of remote cohort proxies
@@ -21,27 +38,108 @@ import java.util.List;
 public class ThreePhaseCommitCohortProxy implements
     DOMStoreThreePhaseCommitCohort{
 
+    private static final Logger
+        LOG = LoggerFactory.getLogger(DistributedDataStore.class);
+
+    private final ActorContext actorContext;
     private final List<ActorPath> cohortPaths;
+    //FIXME : Use a thread pool here
+    private final ExecutorService executorService = Executors.newSingleThreadExecutor();
 
-    public ThreePhaseCommitCohortProxy(List<ActorPath> cohortPaths) {
 
+    public ThreePhaseCommitCohortProxy(ActorContext actorContext, List<ActorPath> cohortPaths) {
+        this.actorContext = actorContext;
         this.cohortPaths = cohortPaths;
     }
 
     @Override public ListenableFuture<Boolean> canCommit() {
-        throw new UnsupportedOperationException("canCommit");
+        Callable<Boolean> call = new Callable() {
+
+            @Override public Boolean call() throws Exception {
+            for(ActorPath actorPath : cohortPaths){
+                ActorSelection cohort = actorContext.actorSelection(actorPath);
+
+                try {
+                    Object response =
+                        actorContext.executeRemoteOperation(cohort,
+                            new CanCommitTransaction(),
+                            ActorContext.ASK_DURATION);
+
+                    if (response instanceof CanCommitTransactionReply) {
+                        CanCommitTransactionReply reply =
+                            (CanCommitTransactionReply) response;
+                        if (!reply.getCanCommit()) {
+                            return false;
+                        }
+                    }
+                } catch(RuntimeException e){
+                    LOG.error("Unexpected Exception", e);
+                    return false;
+                }
+
+
+            }
+            return true;
+            }
+        };
+
+        ListenableFutureTask<Boolean>
+            future = ListenableFutureTask.create(call);
+
+        executorService.submit(future);
+
+        return future;
     }
 
     @Override public ListenableFuture<Void> preCommit() {
-        throw new UnsupportedOperationException("preCommit");
+        return voidOperation(new PreCommitTransaction(), PreCommitTransactionReply.class);
     }
 
     @Override public ListenableFuture<Void> abort() {
-        throw new UnsupportedOperationException("abort");
+        return voidOperation(new AbortTransaction(), AbortTransactionReply.class);
     }
 
     @Override public ListenableFuture<Void> commit() {
-        throw new UnsupportedOperationException("commit");
+        return voidOperation(new CommitTransaction(), CommitTransactionReply.class);
+    }
+
+    private ListenableFuture<Void> voidOperation(final Object message, final Class expectedResponseClass){
+        Callable<Void> call = new Callable<Void>() {
+
+            @Override public Void call() throws Exception {
+                for(ActorPath actorPath : cohortPaths){
+                    ActorSelection cohort = actorContext.actorSelection(actorPath);
+
+                    try {
+                        Object response =
+                            actorContext.executeRemoteOperation(cohort,
+                                message,
+                                ActorContext.ASK_DURATION);
+
+                        if (response != null && !response.getClass()
+                            .equals(expectedResponseClass)) {
+                            throw new RuntimeException(
+                                String.format(
+                                    "did not get the expected response \n\t\t expected : %s \n\t\t actual   : %s",
+                                    expectedResponseClass.toString(),
+                                    response.getClass().toString())
+                            );
+                        }
+                    } catch(TimeoutException e){
+                        LOG.error(String.format("A timeout occurred when processing operation : %s", message));
+                    }
+                }
+                return null;
+            }
+        };
+
+        ListenableFutureTask<Void>
+            future = ListenableFutureTask.create(call);
+
+        executorService.submit(future);
+
+        return future;
+
     }
 
     public List<ActorPath> getCohortPaths() {
index 837ffc1..91e903f 100644 (file)
@@ -44,6 +44,7 @@ public class TransactionChainProxy implements DOMStoreTransactionChain{
 
     @Override
     public void close() {
+        // FIXME : The problem here is don't know which shard the transaction chain is to be created on ???
         throw new UnsupportedOperationException("close - not sure what to do here?");
     }
 }
index 32bb7d0..c122761 100644 (file)
@@ -59,17 +59,17 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction {
 
     private static final AtomicLong counter = new AtomicLong();
 
-    private final TransactionType readOnly;
+    private final TransactionType transactionType;
     private final ActorContext actorContext;
     private final Map<String, ActorSelection> remoteTransactionPaths = new HashMap<>();
     private final String identifier;
 
     public TransactionProxy(
         ActorContext actorContext,
-        TransactionType readOnly) {
+        TransactionType transactionType) {
 
         this.identifier = "transaction-" + counter.getAndIncrement();
-        this.readOnly = readOnly;
+        this.transactionType = transactionType;
         this.actorContext = actorContext;
 
         Object response = actorContext.executeShardOperation(Shard.DEFAULT_NAME, new CreateTransaction(), ActorContext.ASK_DURATION);
@@ -142,7 +142,7 @@ public class TransactionProxy implements DOMStoreReadWriteTransaction {
             }
         }
 
-        return new ThreePhaseCommitCohortProxy(cohortPaths);
+        return new ThreePhaseCommitCohortProxy(actorContext, cohortPaths);
     }
 
     @Override
diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/exceptions/PrimaryNotFoundException.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/exceptions/PrimaryNotFoundException.java
new file mode 100644 (file)
index 0000000..5a131ad
--- /dev/null
@@ -0,0 +1,12 @@
+/*
+ * Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+
+package org.opendaylight.controller.cluster.datastore.exceptions;
+
+public class PrimaryNotFoundException extends RuntimeException {
+}
diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/exceptions/TimeoutException.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/exceptions/TimeoutException.java
new file mode 100644 (file)
index 0000000..4780aac
--- /dev/null
@@ -0,0 +1,15 @@
+/*
+ * Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+
+package org.opendaylight.controller.cluster.datastore.exceptions;
+
+public class TimeoutException extends RuntimeException {
+    public TimeoutException(Exception e){
+        super(e);
+    }
+}
index ba4d4de..0aa205f 100644 (file)
@@ -13,6 +13,8 @@ import akka.actor.ActorRef;
 import akka.actor.ActorSelection;
 import akka.actor.ActorSystem;
 import akka.util.Timeout;
+import org.opendaylight.controller.cluster.datastore.exceptions.PrimaryNotFoundException;
+import org.opendaylight.controller.cluster.datastore.exceptions.TimeoutException;
 import org.opendaylight.controller.cluster.datastore.messages.FindPrimary;
 import org.opendaylight.controller.cluster.datastore.messages.PrimaryFound;
 import org.slf4j.Logger;
@@ -81,7 +83,7 @@ public class ActorContext {
 
             return actorSystem.actorSelection(found.getPrimaryPath());
         }
-        throw new RuntimeException("primary was not found");
+        throw new PrimaryNotFoundException();
     }
 
     /**
@@ -99,7 +101,7 @@ public class ActorContext {
         try {
             return Await.result(future, AWAIT_DURATION);
         } catch (Exception e) {
-            throw new RuntimeException(e);
+            throw new TimeoutException(e);
         }
     }
 
@@ -118,7 +120,7 @@ public class ActorContext {
         try {
             return Await.result(future, AWAIT_DURATION);
         } catch (Exception e) {
-            throw new RuntimeException(e);
+            throw new TimeoutException(e);
         }
     }
 
@@ -131,7 +133,8 @@ public class ActorContext {
      * @param shardName
      * @param message
      * @param duration
-     * @throws java.lang.RuntimeException when a primary is not found or if the message to the remote shard fails or times out
+     * @throws org.opendaylight.controller.cluster.datastore.exceptions.TimeoutException if the message to the remote shard times out
+     * @throws org.opendaylight.controller.cluster.datastore.exceptions.PrimaryNotFoundException if the primary shard is not found
      *
      * @return
      */
diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/config/yang/config/distributed_datastore_provider/DistributedConfigDataStoreProviderModule.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/config/yang/config/distributed_datastore_provider/DistributedConfigDataStoreProviderModule.java
new file mode 100644 (file)
index 0000000..ad2fb02
--- /dev/null
@@ -0,0 +1,40 @@
+package org.opendaylight.controller.config.yang.config.distributed_datastore_provider;
+
+import akka.actor.ActorSystem;
+import com.typesafe.config.ConfigFactory;
+import org.opendaylight.controller.cluster.datastore.DistributedDataStore;
+
+public class DistributedConfigDataStoreProviderModule extends org.opendaylight.controller.config.yang.config.distributed_datastore_provider.AbstractDistributedConfigDataStoreProviderModule {
+    public DistributedConfigDataStoreProviderModule(org.opendaylight.controller.config.api.ModuleIdentifier identifier, org.opendaylight.controller.config.api.DependencyResolver dependencyResolver) {
+        super(identifier, dependencyResolver);
+    }
+
+    public DistributedConfigDataStoreProviderModule(org.opendaylight.controller.config.api.ModuleIdentifier identifier, org.opendaylight.controller.config.api.DependencyResolver dependencyResolver, org.opendaylight.controller.config.yang.config.distributed_datastore_provider.DistributedConfigDataStoreProviderModule oldModule, java.lang.AutoCloseable oldInstance) {
+        super(identifier, dependencyResolver, oldModule, oldInstance);
+    }
+
+    @Override
+    public void customValidation() {
+        // add custom validation form module attributes here.
+    }
+
+    @Override
+    public java.lang.AutoCloseable createInstance() {
+     final ActorSystem actorSystem = ActorSystem.create("opendaylight-cluster-system", ConfigFactory
+          .load().getConfig("ODLCluster"));
+
+
+      final DistributedDataStore configDatastore = new DistributedDataStore(actorSystem, "config");
+
+      final class AutoCloseableDistributedDataStore implements AutoCloseable {
+
+        @Override
+        public void close() throws Exception {
+          actorSystem.shutdown();
+        }
+      }
+
+      return new AutoCloseableDistributedDataStore();
+    }
+
+}
@@ -1,13 +1,13 @@
 /*
 * Generated file
 *
-* Generated from: yang module name: distributed-datastore-provider yang module local name: distributed-datastore-provider
+* Generated from: yang module name: distributed-datastore-provider yang module local name: distributed-config-datastore-provider
 * Generated by: org.opendaylight.controller.config.yangjmxgenerator.plugin.JMXGenerator
-* Generated at: Thu Jun 12 15:23:43 PDT 2014
+* Generated at: Tue Jun 24 17:14:50 PDT 2014
 *
 * Do not modify this file unless it is present under src/main directory
 */
 package org.opendaylight.controller.config.yang.config.distributed_datastore_provider;
-public class DistributedDataStoreProviderModuleFactory extends org.opendaylight.controller.config.yang.config.distributed_datastore_provider.AbstractDistributedDataStoreProviderModuleFactory {
+public class DistributedConfigDataStoreProviderModuleFactory extends org.opendaylight.controller.config.yang.config.distributed_datastore_provider.AbstractDistributedConfigDataStoreProviderModuleFactory {
 
 }
diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/config/yang/config/distributed_datastore_provider/DistributedDataStoreProviderModule.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/config/yang/config/distributed_datastore_provider/DistributedDataStoreProviderModule.java
deleted file mode 100644 (file)
index 3a78f93..0000000
+++ /dev/null
@@ -1,36 +0,0 @@
-package org.opendaylight.controller.config.yang.config.distributed_datastore_provider;
-
-import akka.actor.ActorSystem;
-import org.opendaylight.controller.cluster.datastore.DistributedDataStore;
-
-public class DistributedDataStoreProviderModule extends org.opendaylight.controller.config.yang.config.distributed_datastore_provider.AbstractDistributedDataStoreProviderModule {
-  public DistributedDataStoreProviderModule(org.opendaylight.controller.config.api.ModuleIdentifier identifier, org.opendaylight.controller.config.api.DependencyResolver dependencyResolver) {
-    super(identifier, dependencyResolver);
-  }
-
-  public DistributedDataStoreProviderModule(org.opendaylight.controller.config.api.ModuleIdentifier identifier, org.opendaylight.controller.config.api.DependencyResolver dependencyResolver, org.opendaylight.controller.config.yang.config.distributed_datastore_provider.DistributedDataStoreProviderModule oldModule, java.lang.AutoCloseable oldInstance) {
-    super(identifier, dependencyResolver, oldModule, oldInstance);
-  }
-
-  @Override
-  public void customValidation() {
-    // add custom validation form module attributes here.
-  }
-
-  @Override
-  public java.lang.AutoCloseable createInstance() {
-    ActorSystem actorSystem = ActorSystem.create("opendaylight-cluster");
-    final DistributedDataStore configurationStore = new DistributedDataStore(actorSystem, "config");
-    final DistributedDataStore operationalStore = new DistributedDataStore(actorSystem, "operational");
-
-    final class AutoCloseableDistributedDataStore implements AutoCloseable {
-
-      @Override
-      public void close() throws Exception {
-      }
-    }
-
-    return new AutoCloseableDistributedDataStore();
-  }
-
-}
diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/config/yang/config/distributed_datastore_provider/DistributedOperationalDataStoreProviderModule.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/config/yang/config/distributed_datastore_provider/DistributedOperationalDataStoreProviderModule.java
new file mode 100644 (file)
index 0000000..9e215a4
--- /dev/null
@@ -0,0 +1,38 @@
+package org.opendaylight.controller.config.yang.config.distributed_datastore_provider;
+
+import akka.actor.ActorSystem;
+import com.typesafe.config.ConfigFactory;
+import org.opendaylight.controller.cluster.datastore.DistributedDataStore;
+
+public class DistributedOperationalDataStoreProviderModule extends org.opendaylight.controller.config.yang.config.distributed_datastore_provider.AbstractDistributedOperationalDataStoreProviderModule {
+    public DistributedOperationalDataStoreProviderModule(org.opendaylight.controller.config.api.ModuleIdentifier identifier, org.opendaylight.controller.config.api.DependencyResolver dependencyResolver) {
+        super(identifier, dependencyResolver);
+    }
+
+    public DistributedOperationalDataStoreProviderModule(org.opendaylight.controller.config.api.ModuleIdentifier identifier, org.opendaylight.controller.config.api.DependencyResolver dependencyResolver, org.opendaylight.controller.config.yang.config.distributed_datastore_provider.DistributedOperationalDataStoreProviderModule oldModule, java.lang.AutoCloseable oldInstance) {
+        super(identifier, dependencyResolver, oldModule, oldInstance);
+    }
+
+    @Override
+    public void customValidation() {
+        // add custom validation form module attributes here.
+    }
+
+  @Override
+  public java.lang.AutoCloseable createInstance() {
+    final ActorSystem actorSystem = ActorSystem.create("opendaylight-cluster", ConfigFactory
+        .load().getConfig("ODLCluster"));
+    final DistributedDataStore operationalStore = new DistributedDataStore(actorSystem, "operational");
+
+    final class AutoCloseableDistributedDataStore implements AutoCloseable {
+
+      @Override
+      public void close() throws Exception {
+        actorSystem.shutdown();
+      }
+    }
+
+    return new AutoCloseableDistributedDataStore();
+  }
+
+}
diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/config/yang/config/distributed_datastore_provider/DistributedOperationalDataStoreProviderModuleFactory.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/config/yang/config/distributed_datastore_provider/DistributedOperationalDataStoreProviderModuleFactory.java
new file mode 100644 (file)
index 0000000..c9965fe
--- /dev/null
@@ -0,0 +1,13 @@
+/*
+* Generated file
+*
+* Generated from: yang module name: distributed-datastore-provider yang module local name: distributed-operational-datastore-provider
+* Generated by: org.opendaylight.controller.config.yangjmxgenerator.plugin.JMXGenerator
+* Generated at: Tue Jun 24 17:14:50 PDT 2014
+*
+* Do not modify this file unless it is present under src/main directory
+*/
+package org.opendaylight.controller.config.yang.config.distributed_datastore_provider;
+public class DistributedOperationalDataStoreProviderModuleFactory extends org.opendaylight.controller.config.yang.config.distributed_datastore_provider.AbstractDistributedOperationalDataStoreProviderModuleFactory {
+
+}
diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/resources/application.conf b/opendaylight/md-sal/sal-distributed-datastore/src/main/resources/application.conf
new file mode 100644 (file)
index 0000000..b56c26b
--- /dev/null
@@ -0,0 +1,3 @@
+ODLCluster{
+
+}
\ No newline at end of file
index 0471bc8..ab824c4 100644 (file)
@@ -7,6 +7,9 @@ module distributed-datastore-provider {
 
     import config { prefix config; revision-date 2013-04-05; }
     import rpc-context { prefix rpcx; revision-date 2013-06-17; }
+    import opendaylight-config-dom-datastore {prefix config-dom-store-spi;}
+    import opendaylight-operational-dom-datastore {prefix operational-dom-store-spi;}
+    import opendaylight-md-sal-dom {prefix sal;}
 
     description
         "This module contains the base YANG definitions for
@@ -18,17 +21,32 @@ module distributed-datastore-provider {
     }
 
     // This is the definition of the service implementation as a module identity.
-    identity distributed-datastore-provider {
+    identity distributed-config-datastore-provider {
             base config:module-type;
-
+            config:provided-service config-dom-store-spi:config-dom-datastore;
             // Specifies the prefix for generated java classes.
-            config:java-name-prefix DistributedDataStoreProvider;
+            config:java-name-prefix DistributedConfigDataStoreProvider;
     }
 
+     // This is the definition of the service implementation as a module identity.
+     identity distributed-operational-datastore-provider {
+                base config:module-type;
+                config:provided-service operational-dom-store-spi:operational-dom-datastore;
+                // Specifies the prefix for generated java classes.
+                config:java-name-prefix DistributedOperationalDataStoreProvider;
+     }
+
     // Augments the 'configuration' choice node under modules/module.
     augment "/config:modules/config:module/config:configuration" {
-        case distributed-datastore-provider {
-            when "/config:modules/config:module/config:type = 'distributed-datastore-provider'";
+        case distributed-config-datastore-provider {
+            when "/config:modules/config:module/config:type = 'distributed-config-datastore-provider'";
         }
     }
+
+    // Augments the 'configuration' choice node under modules/module.
+        augment "/config:modules/config:module/config:configuration" {
+            case distributed-operational-datastore-provider {
+                when "/config:modules/config:module/config:type = 'distributed-operational-datastore-provider'";
+            }
+        }
 }
index 8c3ec82..74c858e 100644 (file)
@@ -12,6 +12,7 @@ import akka.actor.ActorPath;
 import akka.actor.ActorRef;
 import akka.actor.ActorSelection;
 import akka.actor.Props;
+import akka.actor.Terminated;
 import akka.testkit.JavaTestKit;
 import junit.framework.Assert;
 import org.junit.Test;
@@ -30,11 +31,14 @@ import org.opendaylight.controller.cluster.datastore.messages.WriteData;
 import org.opendaylight.controller.cluster.datastore.messages.WriteDataReply;
 import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
 import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
+import scala.concurrent.Await;
+import scala.concurrent.Future;
+import scala.concurrent.duration.FiniteDuration;
 
 public class BasicIntegrationTest extends AbstractActorTest {
 
     @Test
-    public void integrationTest() {
+    public void integrationTest() throws Exception{
         // This test will
         // - create a Shard
         // - initiate a transaction
@@ -57,7 +61,7 @@ public class BasicIntegrationTest extends AbstractActorTest {
                     shard.tell(new CreateTransactionChain(), getRef());
 
                     final ActorSelection transactionChain =
-                        new ExpectMsg<ActorSelection>("match hint") {
+                        new ExpectMsg<ActorSelection>("CreateTransactionChainReply") {
                             protected ActorSelection match(Object in) {
                                 if (in instanceof CreateTransactionChainReply) {
                                     ActorPath transactionChainPath =
@@ -76,7 +80,7 @@ public class BasicIntegrationTest extends AbstractActorTest {
                     transactionChain.tell(new CreateTransaction(), getRef());
 
                     final ActorSelection transaction =
-                        new ExpectMsg<ActorSelection>("match hint") {
+                        new ExpectMsg<ActorSelection>("CreateTransactionReply") {
                             protected ActorSelection match(Object in) {
                                 if (in instanceof CreateTransactionReply) {
                                     ActorPath transactionPath =
@@ -92,11 +96,14 @@ public class BasicIntegrationTest extends AbstractActorTest {
 
                     Assert.assertNotNull(transaction);
 
+                    // Add a watch on the transaction actor so that we are notified when it dies
+                    final ActorRef transactionActorRef = watchActor(transaction);
+
                     transaction.tell(new WriteData(TestModel.TEST_PATH,
                         ImmutableNodes.containerNode(TestModel.TEST_QNAME)),
                         getRef());
 
-                    Boolean writeDone = new ExpectMsg<Boolean>("match hint") {
+                    Boolean writeDone = new ExpectMsg<Boolean>("WriteDataReply") {
                         protected Boolean match(Object in) {
                             if (in instanceof WriteDataReply) {
                                 return true;
@@ -111,7 +118,7 @@ public class BasicIntegrationTest extends AbstractActorTest {
                     transaction.tell(new ReadyTransaction(), getRef());
 
                     final ActorSelection cohort =
-                        new ExpectMsg<ActorSelection>("match hint") {
+                        new ExpectMsg<ActorSelection>("ReadyTransactionReply") {
                             protected ActorSelection match(Object in) {
                                 if (in instanceof ReadyTransactionReply) {
                                     ActorPath cohortPath =
@@ -127,10 +134,13 @@ public class BasicIntegrationTest extends AbstractActorTest {
 
                     Assert.assertNotNull(cohort);
 
+                    // Add a watch on the transaction actor so that we are notified when it dies
+                    final ActorRef cohorActorRef = watchActor(cohort);
+
                     cohort.tell(new PreCommitTransaction(), getRef());
 
                     Boolean preCommitDone =
-                        new ExpectMsg<Boolean>("match hint") {
+                        new ExpectMsg<Boolean>("PreCommitTransactionReply") {
                             protected Boolean match(Object in) {
                                 if (in instanceof PreCommitTransactionReply) {
                                     return true;
@@ -144,8 +154,35 @@ public class BasicIntegrationTest extends AbstractActorTest {
 
                     cohort.tell(new CommitTransaction(), getRef());
 
+                    final Boolean terminatedCohort =
+                        new ExpectMsg<Boolean>("Terminated Cohort") {
+                            protected Boolean match(Object in) {
+                                if (in instanceof Terminated) {
+                                    return cohorActorRef.equals(((Terminated) in).actor());
+                                } else {
+                                    throw noMatch();
+                                }
+                            }
+                        }.get(); // this extracts the received message
+
+                    Assert.assertTrue(terminatedCohort);
+
+
+                    final Boolean terminatedTransaction =
+                        new ExpectMsg<Boolean>("Terminated Transaction") {
+                            protected Boolean match(Object in) {
+                                if (in instanceof Terminated) {
+                                    return transactionActorRef.equals(((Terminated) in).actor());
+                                } else {
+                                    throw noMatch();
+                                }
+                            }
+                        }.get(); // this extracts the received message
+
+                    Assert.assertTrue(terminatedTransaction);
+
                     final Boolean commitDone =
-                        new ExpectMsg<Boolean>("match hint") {
+                        new ExpectMsg<Boolean>("CommitTransactionReply") {
                             protected Boolean match(Object in) {
                                 if (in instanceof CommitTransactionReply) {
                                     return true;
@@ -161,7 +198,25 @@ public class BasicIntegrationTest extends AbstractActorTest {
 
 
             };
-        }};
+        }
+
+            private ActorRef watchActor(ActorSelection actor) {
+                Future<ActorRef> future = actor
+                    .resolveOne(FiniteDuration.apply(100, "milliseconds"));
+
+                try {
+                    ActorRef actorRef = Await.result(future,
+                        FiniteDuration.apply(100, "milliseconds"));
+
+                    watch(actorRef);
+
+                    return actorRef;
+                } catch (Exception e) {
+                    throw new RuntimeException(e);
+                }
+
+            }
+        };
 
 
     }
diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/DistributedDataStoreIntegrationTest.java b/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/DistributedDataStoreIntegrationTest.java
new file mode 100644 (file)
index 0000000..f400e74
--- /dev/null
@@ -0,0 +1,54 @@
+package org.opendaylight.controller.cluster.datastore;
+
+import com.google.common.base.Optional;
+import com.google.common.util.concurrent.ListenableFuture;
+import org.junit.Test;
+import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
+import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction;
+import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
+import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
+import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
+
+import static junit.framework.Assert.assertEquals;
+import static junit.framework.Assert.assertTrue;
+
+public class DistributedDataStoreIntegrationTest extends AbstractActorTest {
+
+    @Test
+    public void integrationTest() throws Exception {
+        DistributedDataStore distributedDataStore =
+            new DistributedDataStore(getSystem(), "config");
+
+        distributedDataStore.onGlobalContextUpdated(TestModel.createTestContext());
+
+        DOMStoreReadWriteTransaction transaction =
+            distributedDataStore.newReadWriteTransaction();
+
+        transaction.write(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
+
+        ListenableFuture<Optional<NormalizedNode<?, ?>>> future =
+            transaction.read(TestModel.TEST_PATH);
+
+        Optional<NormalizedNode<?, ?>> optional = future.get();
+
+        NormalizedNode<?, ?> normalizedNode = optional.get();
+
+        assertEquals(TestModel.TEST_QNAME, normalizedNode.getNodeType());
+
+        DOMStoreThreePhaseCommitCohort ready = transaction.ready();
+
+        ListenableFuture<Boolean> canCommit = ready.canCommit();
+
+        assertTrue(canCommit.get());
+
+        ListenableFuture<Void> preCommit = ready.preCommit();
+
+        preCommit.get();
+
+        ListenableFuture<Void> commit = ready.commit();
+
+        commit.get();
+
+    }
+
+}
index 48365fa..b9ab8a3 100644 (file)
@@ -4,8 +4,10 @@ import akka.actor.ActorRef;
 import akka.actor.Props;
 import akka.testkit.JavaTestKit;
 import org.junit.Test;
+import org.opendaylight.controller.cluster.datastore.messages.CreateTransaction;
 import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionChain;
 import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionChainReply;
+import org.opendaylight.controller.cluster.datastore.messages.CreateTransactionReply;
 import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListener;
 import org.opendaylight.controller.cluster.datastore.messages.RegisterChangeListenerReply;
 import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContext;
@@ -18,83 +20,138 @@ import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
 
 import static org.junit.Assert.assertTrue;
 
-public class ShardTest extends AbstractActorTest{
-  @Test
-  public void testOnReceiveCreateTransactionChain() throws Exception {
-    new JavaTestKit(getSystem()) {{
-      final Props props = Shard.props("config");
-      final ActorRef subject = getSystem().actorOf(props, "testCreateTransactionChain");
-
-      new Within(duration("1 seconds")) {
-        protected void run() {
-
-          subject.tell(new CreateTransactionChain(), getRef());
-
-          final String out = new ExpectMsg<String>("match hint") {
-            // do not put code outside this method, will run afterwards
-            protected String match(Object in) {
-              if (in instanceof CreateTransactionChainReply) {
-                CreateTransactionChainReply reply = (CreateTransactionChainReply) in;
-                return reply.getTransactionChainPath().toString();
-              } else {
-                throw noMatch();
-              }
-            }
-          }.get(); // this extracts the received message
-
-          assertTrue(out.matches("akka:\\/\\/test\\/user\\/testCreateTransactionChain\\/\\$.*"));
-          // Will wait for the rest of the 3 seconds
-          expectNoMsg();
-        }
-
-
-      };
-    }};
-  }
-
-  @Test
-  public void testOnReceiveRegisterListener() throws Exception {
-    new JavaTestKit(getSystem()) {{
-      final Props props = Shard.props("config");
-      final ActorRef subject = getSystem().actorOf(props, "testRegisterChangeListener");
-
-      new Within(duration("1 seconds")) {
-        protected void run() {
-
-          subject.tell(new UpdateSchemaContext(TestModel.createTestContext()), getRef());
-
-          subject.tell(new RegisterChangeListener(TestModel.TEST_PATH, getRef().path() , AsyncDataBroker.DataChangeScope.BASE), getRef());
+public class ShardTest extends AbstractActorTest {
+    @Test
+    public void testOnReceiveCreateTransactionChain() throws Exception {
+        new JavaTestKit(getSystem()) {{
+            final Props props = Shard.props("config");
+            final ActorRef subject =
+                getSystem().actorOf(props, "testCreateTransactionChain");
+
+            new Within(duration("1 seconds")) {
+                protected void run() {
+
+                    subject.tell(new CreateTransactionChain(), getRef());
+
+                    final String out = new ExpectMsg<String>("match hint") {
+                        // do not put code outside this method, will run afterwards
+                        protected String match(Object in) {
+                            if (in instanceof CreateTransactionChainReply) {
+                                CreateTransactionChainReply reply =
+                                    (CreateTransactionChainReply) in;
+                                return reply.getTransactionChainPath()
+                                    .toString();
+                            } else {
+                                throw noMatch();
+                            }
+                        }
+                    }.get(); // this extracts the received message
+
+                    assertTrue(out.matches(
+                        "akka:\\/\\/test\\/user\\/testCreateTransactionChain\\/\\$.*"));
+                    // Will wait for the rest of the 3 seconds
+                    expectNoMsg();
+                }
+
+
+            };
+        }};
+    }
+
+    @Test
+    public void testOnReceiveRegisterListener() throws Exception {
+        new JavaTestKit(getSystem()) {{
+            final Props props = Shard.props("config");
+            final ActorRef subject =
+                getSystem().actorOf(props, "testRegisterChangeListener");
+
+            new Within(duration("1 seconds")) {
+                protected void run() {
+
+                    subject.tell(
+                        new UpdateSchemaContext(TestModel.createTestContext()),
+                        getRef());
+
+                    subject.tell(new RegisterChangeListener(TestModel.TEST_PATH,
+                        getRef().path(), AsyncDataBroker.DataChangeScope.BASE),
+                        getRef());
+
+                    final String out = new ExpectMsg<String>("match hint") {
+                        // do not put code outside this method, will run afterwards
+                        protected String match(Object in) {
+                            if (in instanceof RegisterChangeListenerReply) {
+                                RegisterChangeListenerReply reply =
+                                    (RegisterChangeListenerReply) in;
+                                return reply.getListenerRegistrationPath()
+                                    .toString();
+                            } else {
+                                throw noMatch();
+                            }
+                        }
+                    }.get(); // this extracts the received message
+
+                    assertTrue(out.matches(
+                        "akka:\\/\\/test\\/user\\/testRegisterChangeListener\\/\\$.*"));
+                    // Will wait for the rest of the 3 seconds
+                    expectNoMsg();
+                }
+
+
+            };
+        }};
+    }
+
+    @Test
+    public void testCreateTransaction(){
+        new JavaTestKit(getSystem()) {{
+            final Props props = Shard.props("config");
+            final ActorRef subject =
+                getSystem().actorOf(props, "testCreateTransaction");
+
+            new Within(duration("1 seconds")) {
+                protected void run() {
+
+                    subject.tell(
+                        new UpdateSchemaContext(TestModel.createTestContext()),
+                        getRef());
+
+                    subject.tell(new CreateTransaction(),
+                        getRef());
+
+                    final String out = new ExpectMsg<String>("match hint") {
+                        // do not put code outside this method, will run afterwards
+                        protected String match(Object in) {
+                            if (in instanceof CreateTransactionReply) {
+                                CreateTransactionReply reply =
+                                    (CreateTransactionReply) in;
+                                return reply.getTransactionPath()
+                                    .toString();
+                            } else {
+                                throw noMatch();
+                            }
+                        }
+                    }.get(); // this extracts the received message
+
+                    assertTrue(out.matches(
+                        "akka:\\/\\/test\\/user\\/testCreateTransaction\\/\\$.*"));
+                    // Will wait for the rest of the 3 seconds
+                    expectNoMsg();
+                }
+
+
+            };
+        }};
+    }
+
+
+
+    private AsyncDataChangeListener<InstanceIdentifier, NormalizedNode<?, ?>> noOpDataChangeListener() {
+        return new AsyncDataChangeListener<InstanceIdentifier, NormalizedNode<?, ?>>() {
+            @Override
+            public void onDataChanged(
+                AsyncDataChangeEvent<InstanceIdentifier, NormalizedNode<?, ?>> change) {
 
-          final String out = new ExpectMsg<String>("match hint") {
-            // do not put code outside this method, will run afterwards
-            protected String match(Object in) {
-              if (in instanceof RegisterChangeListenerReply) {
-                RegisterChangeListenerReply reply = (RegisterChangeListenerReply) in;
-                return reply.getListenerRegistrationPath().toString();
-              } else {
-                throw noMatch();
-              }
             }
-          }.get(); // this extracts the received message
-
-          assertTrue(out.matches("akka:\\/\\/test\\/user\\/testRegisterChangeListener\\/\\$.*"));
-          // Will wait for the rest of the 3 seconds
-          expectNoMsg();
-        }
-
-
-      };
-    }};
-  }
-
-
-
-  private  AsyncDataChangeListener<InstanceIdentifier, NormalizedNode<?, ?>> noOpDataChangeListener(){
-    return new AsyncDataChangeListener<InstanceIdentifier, NormalizedNode<?, ?>>() {
-      @Override
-      public void onDataChanged(AsyncDataChangeEvent<InstanceIdentifier, NormalizedNode<?, ?>> change) {
-
-      }
-    };
-  }
+        };
+    }
 }
index 9116f24..e4d8e1b 100644 (file)
@@ -2,6 +2,7 @@ package org.opendaylight.controller.cluster.datastore;
 
 import akka.actor.ActorRef;
 import akka.actor.Props;
+import akka.actor.Terminated;
 import akka.testkit.JavaTestKit;
 import com.google.common.util.concurrent.ListeningExecutorService;
 import com.google.common.util.concurrent.MoreExecutors;
@@ -32,246 +33,328 @@ import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 
 public class ShardTransactionTest extends AbstractActorTest {
-  private static ListeningExecutorService storeExecutor = MoreExecutors.listeningDecorator(MoreExecutors.sameThreadExecutor());
+    private static ListeningExecutorService storeExecutor =
+        MoreExecutors.listeningDecorator(MoreExecutors.sameThreadExecutor());
+
+    private static final InMemoryDOMDataStore store =
+        new InMemoryDOMDataStore("OPER", storeExecutor);
+
+    static {
+        store.onGlobalContextUpdated(TestModel.createTestContext());
+    }
+
+    @Test
+    public void testOnReceiveReadData() throws Exception {
+        new JavaTestKit(getSystem()) {{
+            final ActorRef shard = getSystem().actorOf(Shard.props("config"));
+            final Props props =
+                ShardTransaction.props(store.newReadWriteTransaction(), shard);
+            final ActorRef subject = getSystem().actorOf(props, "testReadData");
+
+            new Within(duration("1 seconds")) {
+                protected void run() {
+
+                    subject.tell(
+                        new ReadData(InstanceIdentifier.builder().build()),
+                        getRef());
+
+                    final String out = new ExpectMsg<String>("match hint") {
+                        // do not put code outside this method, will run afterwards
+                        protected String match(Object in) {
+                            if (in instanceof ReadDataReply) {
+                                if (((ReadDataReply) in).getNormalizedNode()
+                                    != null) {
+                                    return "match";
+                                }
+                                return null;
+                            } else {
+                                throw noMatch();
+                            }
+                        }
+                    }.get(); // this extracts the received message
+
+                    assertEquals("match", out);
+
+                    expectNoMsg();
+                }
+
+
+            };
+        }};
+    }
+
+    @Test
+    public void testOnReceiveReadDataWhenDataNotFound() throws Exception {
+        new JavaTestKit(getSystem()) {{
+            final ActorRef shard = getSystem().actorOf(Shard.props("config"));
+            final Props props =
+                ShardTransaction.props(store.newReadWriteTransaction(), shard);
+            final ActorRef subject = getSystem().actorOf(props, "testReadDataWhenDataNotFound");
+
+            new Within(duration("1 seconds")) {
+                protected void run() {
+
+                    subject.tell(
+                        new ReadData(TestModel.TEST_PATH),
+                        getRef());
+
+                    final String out = new ExpectMsg<String>("match hint") {
+                        // do not put code outside this method, will run afterwards
+                        protected String match(Object in) {
+                            if (in instanceof ReadDataReply) {
+                                if (((ReadDataReply) in).getNormalizedNode()
+                                    == null) {
+                                    return "match";
+                                }
+                                return null;
+                            } else {
+                                throw noMatch();
+                            }
+                        }
+                    }.get(); // this extracts the received message
+
+                    assertEquals("match", out);
+
+                    expectNoMsg();
+                }
+
+
+            };
+        }};
+    }
+
+    private void assertModification(final ActorRef subject,
+        final Class<? extends Modification> modificationType) {
+        new JavaTestKit(getSystem()) {{
+            new Within(duration("1 seconds")) {
+                protected void run() {
+                    subject
+                        .tell(new ShardTransaction.GetCompositedModification(),
+                            getRef());
+
+                    final CompositeModification compositeModification =
+                        new ExpectMsg<CompositeModification>("match hint") {
+                            // do not put code outside this method, will run afterwards
+                            protected CompositeModification match(Object in) {
+                                if (in instanceof ShardTransaction.GetCompositeModificationReply) {
+                                    return ((ShardTransaction.GetCompositeModificationReply) in)
+                                        .getModification();
+                                } else {
+                                    throw noMatch();
+                                }
+                            }
+                        }.get(); // this extracts the received message
+
+                    assertTrue(
+                        compositeModification.getModifications().size() == 1);
+                    assertEquals(modificationType,
+                        compositeModification.getModifications().get(0)
+                            .getClass());
+
+                }
+            };
+        }};
+    }
+
+    @Test
+    public void testOnReceiveWriteData() throws Exception {
+        new JavaTestKit(getSystem()) {{
+            final ActorRef shard = getSystem().actorOf(Shard.props("config"));
+            final Props props =
+                ShardTransaction.props(store.newReadWriteTransaction(), shard);
+            final ActorRef subject =
+                getSystem().actorOf(props, "testWriteData");
+
+            new Within(duration("1 seconds")) {
+                protected void run() {
+
+                    subject.tell(new WriteData(TestModel.TEST_PATH,
+                        ImmutableNodes.containerNode(TestModel.TEST_QNAME)),
+                        getRef());
+
+                    final String out = new ExpectMsg<String>("match hint") {
+                        // do not put code outside this method, will run afterwards
+                        protected String match(Object in) {
+                            if (in instanceof WriteDataReply) {
+                                return "match";
+                            } else {
+                                throw noMatch();
+                            }
+                        }
+                    }.get(); // this extracts the received message
+
+                    assertEquals("match", out);
+
+                    assertModification(subject, WriteModification.class);
+                    expectNoMsg();
+                }
+
+
+            };
+        }};
+    }
 
-  private static final InMemoryDOMDataStore store = new InMemoryDOMDataStore("OPER", storeExecutor);
+    @Test
+    public void testOnReceiveMergeData() throws Exception {
+        new JavaTestKit(getSystem()) {{
+            final ActorRef shard = getSystem().actorOf(Shard.props("config"));
+            final Props props =
+                ShardTransaction.props(store.newReadWriteTransaction(), shard);
+            final ActorRef subject =
+                getSystem().actorOf(props, "testMergeData");
 
-  static {
-    store.onGlobalContextUpdated(TestModel.createTestContext());
-  }
+            new Within(duration("1 seconds")) {
+                protected void run() {
 
-  @Test
-  public void testOnReceiveReadData() throws Exception {
-    new JavaTestKit(getSystem()) {{
-      final ActorRef shard = getSystem().actorOf(Shard.props("config"));
-      final Props props = ShardTransaction.props(store.newReadWriteTransaction(), shard);
-      final ActorRef subject = getSystem().actorOf(props, "testReadData");
+                    subject.tell(new MergeData(TestModel.TEST_PATH,
+                        ImmutableNodes.containerNode(TestModel.TEST_QNAME)),
+                        getRef());
 
-      new Within(duration("1 seconds")) {
-        protected void run() {
+                    final String out = new ExpectMsg<String>("match hint") {
+                        // do not put code outside this method, will run afterwards
+                        protected String match(Object in) {
+                            if (in instanceof MergeDataReply) {
+                                return "match";
+                            } else {
+                                throw noMatch();
+                            }
+                        }
+                    }.get(); // this extracts the received message
 
-          subject.tell(new ReadData(InstanceIdentifier.builder().build()), getRef());
+                    assertEquals("match", out);
 
-          final String out = new ExpectMsg<String>("match hint") {
-            // do not put code outside this method, will run afterwards
-            protected String match(Object in) {
-              if (in instanceof ReadDataReply) {
-                if (((ReadDataReply) in).getNormalizedNode() != null) {
-                  return "match";
+                    assertModification(subject, MergeModification.class);
+
+                    expectNoMsg();
                 }
-                return null;
-              } else {
-                throw noMatch();
-              }
-            }
-          }.get(); // this extracts the received message
-
-          assertEquals("match", out);
-
-          expectNoMsg();
-        }
-
-
-      };
-    }};
-  }
-
-  private void assertModification(final ActorRef subject, final Class<? extends Modification> modificationType){
-    new JavaTestKit(getSystem()) {{
-      new Within(duration("1 seconds")) {
-        protected void run() {
-          subject.tell(new ShardTransaction.GetCompositedModification(), getRef());
-
-          final CompositeModification compositeModification = new ExpectMsg<CompositeModification>("match hint") {
-            // do not put code outside this method, will run afterwards
-            protected CompositeModification match(Object in) {
-              if (in instanceof ShardTransaction.GetCompositeModificationReply) {
-                return ((ShardTransaction.GetCompositeModificationReply) in).getModification();
-              } else {
-                throw noMatch();
-              }
-            }
-          }.get(); // this extracts the received message
-
-          assertTrue(compositeModification.getModifications().size() == 1);
-          assertEquals(modificationType, compositeModification.getModifications().get(0).getClass());
-
-        }
-      };
-    }};
-  }
-
-  @Test
-  public void testOnReceiveWriteData() throws Exception {
-    new JavaTestKit(getSystem()) {{
-      final ActorRef shard = getSystem().actorOf(Shard.props("config"));
-      final Props props = ShardTransaction.props(store.newReadWriteTransaction(), shard);
-      final ActorRef subject = getSystem().actorOf(props, "testWriteData");
-
-      new Within(duration("1 seconds")) {
-        protected void run() {
-
-          subject.tell(new WriteData(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME)), getRef());
-
-          final String out = new ExpectMsg<String>("match hint") {
-            // do not put code outside this method, will run afterwards
-            protected String match(Object in) {
-              if (in instanceof WriteDataReply) {
-                return "match";
-              } else {
-                throw noMatch();
-              }
-            }
-          }.get(); // this extracts the received message
-
-          assertEquals("match", out);
-
-          assertModification(subject, WriteModification.class);
-          expectNoMsg();
-        }
-
-
-      };
-    }};
-  }
-
-  @Test
-  public void testOnReceiveMergeData() throws Exception {
-    new JavaTestKit(getSystem()) {{
-      final ActorRef shard = getSystem().actorOf(Shard.props("config"));
-      final Props props = ShardTransaction.props(store.newReadWriteTransaction(), shard);
-      final ActorRef subject = getSystem().actorOf(props, "testMergeData");
-
-      new Within(duration("1 seconds")) {
-        protected void run() {
-
-          subject.tell(new MergeData(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME)), getRef());
-
-          final String out = new ExpectMsg<String>("match hint") {
-            // do not put code outside this method, will run afterwards
-            protected String match(Object in) {
-              if (in instanceof MergeDataReply) {
-                return "match";
-              } else {
-                throw noMatch();
-              }
-            }
-          }.get(); // this extracts the received message
-
-          assertEquals("match", out);
-
-          assertModification(subject, MergeModification.class);
-
-          expectNoMsg();
-        }
-
-
-      };
-    }};
-  }
-
-  @Test
-  public void testOnReceiveDeleteData() throws Exception {
-    new JavaTestKit(getSystem()) {{
-      final ActorRef shard = getSystem().actorOf(Shard.props("config"));
-      final Props props = ShardTransaction.props(store.newReadWriteTransaction(), shard);
-      final ActorRef subject = getSystem().actorOf(props, "testDeleteData");
 
-      new Within(duration("1 seconds")) {
-        protected void run() {
 
-          subject.tell(new DeleteData(TestModel.TEST_PATH), getRef());
+            };
+        }};
+    }
+
+    @Test
+    public void testOnReceiveDeleteData() throws Exception {
+        new JavaTestKit(getSystem()) {{
+            final ActorRef shard = getSystem().actorOf(Shard.props("config"));
+            final Props props =
+                ShardTransaction.props(store.newReadWriteTransaction(), shard);
+            final ActorRef subject =
+                getSystem().actorOf(props, "testDeleteData");
+
+            new Within(duration("1 seconds")) {
+                protected void run() {
+
+                    subject.tell(new DeleteData(TestModel.TEST_PATH), getRef());
+
+                    final String out = new ExpectMsg<String>("match hint") {
+                        // do not put code outside this method, will run afterwards
+                        protected String match(Object in) {
+                            if (in instanceof DeleteDataReply) {
+                                return "match";
+                            } else {
+                                throw noMatch();
+                            }
+                        }
+                    }.get(); // this extracts the received message
 
-          final String out = new ExpectMsg<String>("match hint") {
-            // do not put code outside this method, will run afterwards
-            protected String match(Object in) {
-              if (in instanceof DeleteDataReply) {
-                return "match";
-              } else {
-                throw noMatch();
-              }
-            }
-          }.get(); // this extracts the received message
+                    assertEquals("match", out);
+
+                    assertModification(subject, DeleteModification.class);
+                    expectNoMsg();
+                }
 
-          assertEquals("match", out);
 
-          assertModification(subject, DeleteModification.class);
-          expectNoMsg();
-        }
+            };
+        }};
+    }
 
 
-      };
-    }};
-  }
+    @Test
+    public void testOnReceiveReadyTransaction() throws Exception {
+        new JavaTestKit(getSystem()) {{
+            final ActorRef shard = getSystem().actorOf(Shard.props("config"));
+            final Props props =
+                ShardTransaction.props(store.newReadWriteTransaction(), shard);
+            final ActorRef subject =
+                getSystem().actorOf(props, "testReadyTransaction");
 
+            new Within(duration("1 seconds")) {
+                protected void run() {
 
-  @Test
-  public void testOnReceiveReadyTransaction() throws Exception {
-    new JavaTestKit(getSystem()) {{
-      final ActorRef shard = getSystem().actorOf(Shard.props("config"));
-      final Props props = ShardTransaction.props(store.newReadWriteTransaction(), shard);
-      final ActorRef subject = getSystem().actorOf(props, "testReadyTransaction");
+                    subject.tell(new ReadyTransaction(), getRef());
 
-      new Within(duration("1 seconds")) {
-        protected void run() {
+                    final String out = new ExpectMsg<String>("match hint") {
+                        // do not put code outside this method, will run afterwards
+                        protected String match(Object in) {
+                            if (in instanceof ReadyTransactionReply) {
+                                return "match";
+                            } else {
+                                throw noMatch();
+                            }
+                        }
+                    }.get(); // this extracts the received message
 
-          subject.tell(new ReadyTransaction(), getRef());
+                    assertEquals("match", out);
 
-          final String out = new ExpectMsg<String>("match hint") {
-            // do not put code outside this method, will run afterwards
-            protected String match(Object in) {
-              if (in instanceof ReadyTransactionReply) {
-                return "match";
-              } else {
-                throw noMatch();
-              }
-            }
-          }.get(); // this extracts the received message
+                    expectNoMsg();
+                }
 
-          assertEquals("match", out);
 
-          expectNoMsg();
-        }
+            };
+        }};
 
+    }
 
-      };
-    }};
+    @Test
+    public void testOnReceiveCloseTransaction() throws Exception {
+        new JavaTestKit(getSystem()) {{
+            final ActorRef shard = getSystem().actorOf(Shard.props("config"));
+            final Props props =
+                ShardTransaction.props(store.newReadWriteTransaction(), shard);
+            final ActorRef subject =
+                getSystem().actorOf(props, "testCloseTransaction");
 
-  }
+            watch(subject);
 
-  @Test
-  public void testOnReceiveCloseTransaction() throws Exception {
-    new JavaTestKit(getSystem()) {{
-      final ActorRef shard = getSystem().actorOf(Shard.props("config"));
-      final Props props = ShardTransaction.props(store.newReadWriteTransaction(), shard);
-      final ActorRef subject = getSystem().actorOf(props, "testCloseTransaction");
+            new Within(duration("2 seconds")) {
+                protected void run() {
 
-      new Within(duration("1 seconds")) {
-        protected void run() {
+                    subject.tell(new CloseTransaction(), getRef());
 
-          subject.tell(new CloseTransaction(), getRef());
+                    final String out = new ExpectMsg<String>("match hint") {
+                        // do not put code outside this method, will run afterwards
+                        protected String match(Object in) {
+                            if (in instanceof CloseTransactionReply) {
+                                return "match";
+                            } else {
+                                throw noMatch();
+                            }
+                        }
+                    }.get(); // this extracts the received message
 
-          final String out = new ExpectMsg<String>("match hint") {
-            // do not put code outside this method, will run afterwards
-            protected String match(Object in) {
-              if (in instanceof CloseTransactionReply) {
-                return "match";
-              } else {
-                throw noMatch();
-              }
-            }
-          }.get(); // this extracts the received message
+                    assertEquals("match", out);
 
-          assertEquals("match", out);
+                    final String termination = new ExpectMsg<String>("match hint") {
+                        // do not put code outside this method, will run afterwards
+                        protected String match(Object in) {
+                            if (in instanceof Terminated) {
+                                return "match";
+                            } else {
+                                throw noMatch();
+                            }
+                        }
+                    }.get(); // this extracts the received message
 
-          expectNoMsg();
-        }
+
+                    expectNoMsg();
+                }
 
 
-      };
-    }};
+            };
+        }};
 
-  }
+    }
 
 
-}
\ No newline at end of file
+}
diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/ThreePhaseCommitCohortProxyTest.java b/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/ThreePhaseCommitCohortProxyTest.java
new file mode 100644 (file)
index 0000000..af3da57
--- /dev/null
@@ -0,0 +1,82 @@
+package org.opendaylight.controller.cluster.datastore;
+
+import akka.actor.ActorRef;
+import akka.actor.Props;
+import com.google.common.util.concurrent.ListenableFuture;
+import junit.framework.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.opendaylight.controller.cluster.datastore.messages.AbortTransactionReply;
+import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransactionReply;
+import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply;
+import org.opendaylight.controller.cluster.datastore.messages.PreCommitTransactionReply;
+import org.opendaylight.controller.cluster.datastore.utils.MessageCollectorActor;
+import org.opendaylight.controller.cluster.datastore.utils.MockActorContext;
+
+import java.util.Arrays;
+
+import static org.junit.Assert.assertNotNull;
+
+public class ThreePhaseCommitCohortProxyTest extends AbstractActorTest {
+
+    private ThreePhaseCommitCohortProxy proxy;
+    private Props props;
+    private ActorRef actorRef;
+    private MockActorContext actorContext;
+
+    @Before
+    public void setUp(){
+        props = Props.create(MessageCollectorActor.class);
+        actorRef = getSystem().actorOf(props);
+        actorContext = new MockActorContext(this.getSystem());
+
+        proxy =
+            new ThreePhaseCommitCohortProxy(actorContext,
+                Arrays.asList(actorRef.path()));
+
+    }
+
+    @Test
+    public void testCanCommit() throws Exception {
+        actorContext.setExecuteRemoteOperationResponse(new CanCommitTransactionReply(true));
+
+        ListenableFuture<Boolean> future = proxy.canCommit();
+
+        Assert.assertTrue(future.get().booleanValue());
+
+    }
+
+    @Test
+    public void testPreCommit() throws Exception {
+        actorContext.setExecuteRemoteOperationResponse(new PreCommitTransactionReply());
+
+        ListenableFuture<Void> future = proxy.preCommit();
+
+        future.get();
+
+    }
+
+    @Test
+    public void testAbort() throws Exception {
+        actorContext.setExecuteRemoteOperationResponse(new AbortTransactionReply());
+
+        ListenableFuture<Void> future = proxy.abort();
+
+        future.get();
+
+    }
+
+    @Test
+    public void testCommit() throws Exception {
+        actorContext.setExecuteRemoteOperationResponse(new CommitTransactionReply());
+
+        ListenableFuture<Void> future = proxy.commit();
+
+        future.get();
+    }
+
+    @Test
+    public void testGetCohortPaths() throws Exception {
+        assertNotNull(proxy.getCohortPaths());
+    }
+}
diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/utils/TestUtils.java b/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/utils/TestUtils.java
new file mode 100644 (file)
index 0000000..bb881d5
--- /dev/null
@@ -0,0 +1,37 @@
+/*
+ * Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+
+package org.opendaylight.controller.cluster.datastore.utils;
+
+import akka.actor.ActorRef;
+import akka.actor.ActorSystem;
+import akka.actor.Props;
+import junit.framework.Assert;
+
+import java.util.List;
+
+public class TestUtils {
+
+    public static void assertFirstSentMessage(ActorSystem actorSystem, ActorRef actorRef, Class clazz){
+        ActorContext testContext = new ActorContext(actorSystem, actorSystem.actorOf(
+            Props.create(DoNothingActor.class)));
+        Object messages = testContext
+            .executeLocalOperation(actorRef, "messages",
+                ActorContext.ASK_DURATION);
+
+        Assert.assertNotNull(messages);
+
+        Assert.assertTrue(messages instanceof List);
+
+        List<Object> listMessages = (List<Object>) messages;
+
+        Assert.assertEquals(1, listMessages.size());
+
+        Assert.assertTrue(listMessages.get(0).getClass().equals(clazz));
+    }
+}
index 0ae33b8..29a27e2 100644 (file)
@@ -48,9 +48,13 @@ public class StatisticsRequestScheduler implements DataTransactionListener {
     private final TimerTask task = new TimerTask() {
         @Override
         public void run() {
-            long now = System.nanoTime();
-            if(now > lastRequestTime+TimeUnit.MILLISECONDS.toNanos(REQUEST_MONITOR_INTERVAL)){
-                requestStatistics();
+            try{
+                long now = System.nanoTime();
+                if(now > lastRequestTime+TimeUnit.MILLISECONDS.toNanos(REQUEST_MONITOR_INTERVAL)){
+                    requestStatistics();
+                }
+            }catch (IllegalArgumentException | IllegalStateException | NullPointerException e){
+                srsLogger.warn("Exception occured while sending statistics request : {}",e);
             }
         }
     };