diff --git a/.gitignore b/.gitignore
index 41f0f27761f8b196b534bf346961c00fcb5a294f..70229e1827bee5c412e2c82ccef3f632b8c91b41 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1,3 +1,5 @@
+.scannerwork
+asan_leak_suppression.txt
ccs_error
ccs_autogen
cov_html
diff --git a/.gitlab-ci.yml b/.gitlab-ci.yml
index fc94b862265bf71e844659259e24fbb701403ec8..e4e8e93e980b8b75449ffb050346bfb76c61fe95 100644
--- a/.gitlab-ci.yml
+++ b/.gitlab-ci.yml
@@ -15,40 +15,19 @@ before_script:
- export CHAOS_LIB_HASH=`git log -n 1 --pretty="%h"`;
- export HEAPCHECK=strict
-# codequality:
-# image: docker:stable
-# stage: quality
-# variables:
-# DOCKER_DRIVER: overlay2
-# allow_failure: true
-# services:
-# - docker:stable-dind
-# script:
-# - export SP_VERSION=$(echo "$CI_SERVER_VERSION" | sed 's/^\([0-9]*\)\.\([0-9]*\).*/\1-\2-stable/')
-# - docker run --env SOURCE_CODE="$PWD" --volume "$PWD"/chaos:/code --volume /var/run/docker.sock:/var/run/docker.sock "registry.gitlab.com/gitlab-org/security-products/codequality:$SP_VERSION" /code
-# artifacts:
-# paths: [codeclimate.json]
-# only:
-# - tags
-# - web
-
-build_framework_x86_64:
+build_ubuntu_14_04_gcc48:
tags:
- chaos
- docker
- linux
- fast
stage: build
- image: baltig.infn.it:4567/bisegni/chaos-docker-compilation:latest
+ image: baltig.infn.it:4567/bisegni/chaos-docker-compilation/ubuntu/14:gcc
script:
- tools/chaos_clean.sh
- cmake -DCHAOS_ARCHITECTURE_TEST=ON -DUSE_GPT=ON -DCMAKE_BUILD_TYPE=PROFILE .
- make clean
- - make -j $NPROC
- - make install
- - cd ccs
- - /Qt/5.9.3/gcc_64/bin/qmake
- - make -j $NPROC
+ - /tmp/build-wrapper/build-wrapper-linux-x86-64 --out-dir bw-output make -j $NPROC install
cache:
key: x86_64_$CI_BUILD_REF_NAME
paths:
@@ -56,32 +35,49 @@ build_framework_x86_64:
artifacts:
paths:
- chaos
+ - ChaosMetadataService
+ - chaos_metadata_service_client
+ - chaos_micro_unit_toolkit
+ - chaos_service_common
+ - ccs
- CHAOSFrameworkTests
+ - bw-output
- 'chaos-distrib-x86_64-Linux'
expire_in: 8 hour
only:
- tags
- web
+ retry: 2
-build_framework_x86_64_test:
+build_ubuntu_14_04_gcc48_test:
dependencies:
- - build_framework_x86_64
+ - build_ubuntu_14_04_gcc48
tags:
- chaos
- docker
- linux
- fast
stage: test
- image: baltig.infn.it:4567/bisegni/chaos-docker-compilation:latest
+ image: baltig.infn.it:4567/bisegni/chaos-docker-compilation/ubuntu/14:gcc
script:
- apt-get update
- apt-get -y install lcov
- export PPROF_PATH=chaos-distrib-x86_64-Linux/bin/pprof
- export LD_LIBRARY_PATH=/builds/chaos-lnf-control/chaosframework/chaos-distrib-x86_64-Linux/lib
- export HEAPCHECK=normal
- - chaos-distrib-x86_64-Linux/bin/TestFramework
+ - chaos-distrib-x86_64-Linux/bin/TestFramework --gtest_output=xml:report_test.xml --gtest_repeat=10 --gtest_break_on_failure
- lcov -c -d chaos --output-file coverage.info
- genhtml coverage.info --output-directory coverage_html
+ - /tmp/sonar-scanner/bin/sonar-scanner
+ -Dsonar.branch.name=$CI_COMMIT_REF_NAME
+ -Dsonar.projectKey=chaosframework
+ -Dsonar.organization=bisegni-github
+ -Dsonar.sources=chaos,ChaosMetadataService,chaos_metadata_service_client,chaos_micro_unit_toolkit,chaos_service_common,ccs
+ -Dsonar.cfamily.build-wrapper-output=bw-output
+ -Dsonar.host.url=https://sonarcloud.io
+ -Dsonar.login=369dba58f43885fc41db2f9076a67b36aa5b2584
+ -Dsonar.cfamily.threads=$NPROC
+ -Dsonar.test.reportPath=build/repost_test.xml
artifacts:
paths:
- 'coverage_html'
@@ -90,6 +86,120 @@ build_framework_x86_64_test:
only:
- tags
- web
+ retry: 2
+
+build_ubuntu_16_04_gcc5:
+ tags:
+ - chaos
+ - docker
+ - linux
+ - fast
+ stage: build
+ image: baltig.infn.it:4567/bisegni/chaos-docker-compilation/ubuntu/16:llvm6
+ script:
+ - tools/chaos_clean.sh
+ - cmake -DCHAOS_ARCHITECTURE_TEST=ON -DUSE_GPT=ON .
+ - make clean
+ - make -j $NPROC
+ - make install
+# - cd ccs
+# - /Qt/5.9.3/gcc_64/bin/qmake
+# - make -j $NPROC
+ cache:
+ key: UBUNTU_1604_LLVM7_$CI_BUILD_REF_NAME
+ paths:
+ - /builds/chaos-lnf-control/chaosframework/config/CACHE/
+ artifacts:
+ paths:
+ - chaos
+ - CHAOSFrameworkTests
+ - 'chaos-distrib-x86_64-Linux'
+ expire_in: 8 hour
+ only:
+ - tags
+ - web
+ retry: 2
+
+build_ubuntu_16_04_gcc5_test:
+ dependencies:
+ - build_ubuntu_16_04_gcc5
+ tags:
+ - chaos
+ - docker
+ - linux
+ - fast
+ stage: test
+ image: baltig.infn.it:4567/bisegni/chaos-docker-compilation/ubuntu/16:llvm6
+ script:
+ - export PPROF_PATH=chaos-distrib-x86_64-Linux/bin/pprof
+ - export LD_LIBRARY_PATH=/builds/chaos-lnf-control/chaosframework/chaos-distrib-x86_64-Linux/lib
+ - export HEAPCHECK=normal
+ - chaos-distrib-x86_64-Linux/bin/TestFramework --gtest_repeat=10 --gtest_break_on_failure
+ artifacts:
+ paths:
+ - 'chaos-distrib-x86_64-Linux'
+ expire_in: 8 hour
+ only:
+ - tags
+ - web
+ retry: 2
+
+build_ubuntu_18_04_gcc73:
+ tags:
+ - chaos
+ - docker
+ - linux
+ - fast
+ stage: build
+ image: baltig.infn.it:4567/bisegni/chaos-docker-compilation/ubuntu/18:gcc7.3
+ script:
+ - tools/chaos_clean.sh
+ - cmake -DCHAOS_ARCHITECTURE_TEST=ON -DUSE_GPT=ON .
+ - make clean
+ - make -j $NPROC
+ - make install
+# - cd ccs
+# - /Qt/5.9.3/gcc_64/bin/qmake
+# - make -j $NPROC
+ cache:
+ key: UBUNTU_1604_LLVM7_$CI_BUILD_REF_NAME
+ paths:
+ - /builds/chaos-lnf-control/chaosframework/config/CACHE/
+ artifacts:
+ paths:
+ - chaos
+ - CHAOSFrameworkTests
+ - 'chaos-distrib-x86_64-Linux'
+ expire_in: 8 hour
+ only:
+ - tags
+ - web
+ retry: 2
+
+build_ubuntu_18_04_gcc73_test:
+ dependencies:
+ - build_ubuntu_18_04_gcc73
+ tags:
+ - chaos
+ - docker
+ - linux
+ - fast
+ stage: test
+ image: baltig.infn.it:4567/bisegni/chaos-docker-compilation/ubuntu/18:gcc7.3
+ script:
+ - export PPROF_PATH=chaos-distrib-x86_64-Linux/bin/pprof
+ - export LD_LIBRARY_PATH=/builds/chaos-lnf-control/chaosframework/chaos-distrib-x86_64-Linux/lib
+ - export HEAPCHECK=normal
+ - chaos-distrib-x86_64-Linux/bin/TestFramework --gtest_repeat=10 --gtest_break_on_failure
+ artifacts:
+ paths:
+ - 'chaos-distrib-x86_64-Linux'
+ expire_in: 8 hour
+ only:
+ - tags
+ - web
+ retry: 2
+
build_framework_centos7:
tags:
@@ -98,17 +208,17 @@ build_framework_centos7:
- linux
- fast
stage: build
- image: baltig.infn.it:4567/bisegni/chaos-docker-compilation/centos7:latest
+ image: baltig.infn.it:4567/bisegni/chaos-docker-compilation/centos7:devtools7
script:
- tools/chaos_clean.sh
- cmake -DCHAOS_VERSION_MAJOR="$CI_COMMIT_REF_NAME.$CHAOS_LIB_HASH" -DCHAOS_VERSION_MINOR="$CI_JOB_STAGE.$CI_COMMIT_SHA" -DCHAOS_BUILD_ID=${CI_PIPELINE_ID} -DCHAOS_ARCHITECTURE_TEST=ON -DUSE_GPT=ON .
- make clean
- make -j $NPROC
- make install
- - cd ccs
- - /Qt/5.9.3/gcc_64/bin/qmake
- - make -j $NPROC
- - mv /builds/chaos-lnf-control/chaosframework/chaos-distrib-x86_64-Linux /builds/chaos-lnf-control/chaosframework/chaos-distrib-x86_64-centos7
+# - cd ccs
+# - /Qt/5.9.3/gcc_64/bin/qmake
+# - make -j $NPROC
+# - mv /builds/chaos-lnf-control/chaosframework/chaos-distrib-x86_64-Linux /builds/chaos-lnf-control/chaosframework/chaos-distrib-x86_64-centos7
cache:
key: x86_64_centos7_$CI_BUILD_REF_NAME
paths:
@@ -117,11 +227,12 @@ build_framework_centos7:
paths:
- chaos
- CHAOSFrameworkTests
- - 'chaos-distrib-x86_64-centos7'
+ - 'chaos-distrib-x86_64-Linux'
expire_in: 8 hour
only:
- tags
- web
+ retry: 2
build_framework_centos7_test:
dependencies:
@@ -132,19 +243,20 @@ build_framework_centos7_test:
- linux
- fast
stage: test
- image: baltig.infn.it:4567/bisegni/chaos-docker-compilation/centos7:latest
+ image: baltig.infn.it:4567/bisegni/chaos-docker-compilation/centos7:devtools7
script:
- - export PPROF_PATH=chaos-distrib-x86_64-centos7/bin/pprof
- - export LD_LIBRARY_PATH=/builds/chaos-lnf-control/chaosframework/chaos-distrib-x86_64-centos7/lib
+ - export PPROF_PATH=chaos-distrib-x86_64-Linux/bin/pprof
+ - export LD_LIBRARY_PATH=/builds/chaos-lnf-control/chaosframework/chaos-distrib-x86_64-Linux/lib
- export HEAPCHECK=normal
- - chaos-distrib-x86_64-centos7/bin/TestFramework
+ - chaos-distrib-x86_64-Linux/bin/TestFramework --gtest_repeat=10 --gtest_break_on_failure
artifacts:
paths:
- - 'chaos-distrib-x86_64-centos7'
+ - 'chaos-distrib-x86_64-Linux'
expire_in: 8 hour
only:
- tags
- web
+ retry: 2
framework_centos7_deploy:
allow_failure: true
@@ -160,7 +272,7 @@ framework_centos7_deploy:
- build_framework_centos7_test
script:
- echo "Prearing deploy for chaos-distrib-x86_64-centos7_$TAR_NAME_POSTFIX.tar.gz"
- - tar cfz chaos-distrib-x86_64-centos7_$TAR_NAME_POSTFIX.tar.gz chaos-distrib-x86_64-centos7
+ - tar cfz chaos-distrib-x86_64-centos7_$TAR_NAME_POSTFIX.tar.gz chaos-distrib-x86_64-Linux
- ls -la
- scp chaos-distrib-x86_64-centos7_$TAR_NAME_POSTFIX.tar.gz chaosweb@opensource.lnf.infn.it:/var/www/html/binary/chaos/$CI_COMMIT_REF_NAME/x86_64/centos/7/chaos-distrib-$TAR_NAME_POSTFIX.tar.gz
- ssh chaosweb@opensource.lnf.infn.it "rm -f /var/www/html/binary/chaos/$CI_COMMIT_REF_NAME/x86_64/centos/7/latest"
@@ -169,52 +281,55 @@ framework_centos7_deploy:
only:
- tags
- web
+ retry: 2
-build_framework_x86_64_c98:
+build_ubuntu_14_04_gcc48_c98:
tags:
- chaos
- docker
- linux
- fast
stage: build
- image: baltig.infn.it:4567/bisegni/chaos-docker-compilation:latest
+ image: baltig.infn.it:4567/bisegni/chaos-docker-compilation/ubuntu/14:gcc
script:
- tools/chaos_clean.sh
- cmake -DCHAOS_ENABLE_C11=OFF -DCHAOS_MDS=OFF -DCHAOS_ARCHITECTURE_TEST=ON -DUSE_GPT=ON -DCMAKE_BUILD_TYPE=PROFILE .
- make clean
- make -j $NPROC
- make install
- - mv /builds/chaos-lnf-control/chaosframework/chaos-distrib-x86_64-Linux /builds/chaos-lnf-control/chaosframework/chaos-distrib-x86_64-Linux-c98
+ # - mv /builds/chaos-lnf-control/chaosframework/chaos-distrib-x86_64-Linux /builds/chaos-lnf-control/chaosframework/chaos-distrib-x86_64-Linux-c98
cache:
key: x86_64_c98_$CI_BUILD_REF_NAME
paths:
- /builds/chaos-lnf-control/chaosframework/config/CACHE/
artifacts:
paths:
- - 'chaos-distrib-x86_64-Linux-c98'
+ - 'chaos-distrib-x86_64-Linux'
expire_in: 8 hour
only:
- tags
- web
+ retry: 2
-build_framework_x86_64_c98_test:
+build_ubuntu_14_04_gcc48_c98_test:
dependencies:
- - build_framework_x86_64_c98
+ - build_ubuntu_14_04_gcc48_c98
tags:
- chaos
- docker
- linux
- fast
stage: test
- image: baltig.infn.it:4567/bisegni/chaos-docker-compilation:latest
+ image: baltig.infn.it:4567/bisegni/chaos-docker-compilation/ubuntu/14:gcc
script:
- - export PPROF_PATH=chaos-distrib-x86_64-Linux-c98/bin/pprof
- - export LD_LIBRARY_PATH=/builds/chaos-lnf-control/chaosframework/chaos-distrib-x86_64-Linux-c98/lib
+ - export PPROF_PATH=chaos-distrib-x86_64-Linux/bin/pprof
+ - export LD_LIBRARY_PATH=/builds/chaos-lnf-control/chaosframework/chaos-distrib-x86_64-Linux/lib
- export HEAPCHECK=normal
- - chaos-distrib-x86_64-Linux-c98/bin/TestFramework
+ - chaos-distrib-x86_64-Linux/bin/TestFramework --gtest_repeat=10 --gtest_break_on_failure
only:
- tags
- web
+ retry: 2
build_framework_arm:
tags:
@@ -241,6 +356,7 @@ build_framework_arm:
only:
- tags
- web
+ retry: 2
llvm_build_framework_x86_64:
tags:
@@ -256,7 +372,7 @@ llvm_build_framework_x86_64:
- make clean
- make -j $NPROC
- make install
- - mv /builds/chaos-lnf-control/chaosframework/chaos-distrib-x86_64-Linux /builds/chaos-lnf-control/chaosframework/chaos-distrib-x86_64-Linux-llvm
+ # - mv /builds/chaos-lnf-control/chaosframework/chaos-distrib-x86_64-Linux /builds/chaos-lnf-control/chaosframework/chaos-distrib-x86_64-Linux-llvm
cache:
key: llvm_x86_64_$CI_BUILD_REF_NAME
paths:
@@ -265,11 +381,12 @@ llvm_build_framework_x86_64:
paths:
- chaos
- CHAOSFrameworkTests
- - 'chaos-distrib-x86_64-Linux-llvm'
+ - 'chaos-distrib-x86_64-Linux'
expire_in: 8 hour
only:
- tags
- web
+ retry: 2
llvm_build_framework_x86_64_test:
tags:
@@ -282,13 +399,14 @@ llvm_build_framework_x86_64_test:
dependencies:
- llvm_build_framework_x86_64
script:
- - export PPROF_PATH=chaos-distrib-x86_64-Linux-llvm/bin/pprof
- - export LD_LIBRARY_PATH=/builds/chaos-lnf-control/chaosframework/chaos-distrib-x86_64-Linux-llvm/lib
+ - export PPROF_PATH=chaos-distrib-x86_64-Linux/bin/pprof
+ - export LD_LIBRARY_PATH=/builds/chaos-lnf-control/chaosframework/chaos-distrib-x86_64-Linux/lib
- export HEAPCHECK=normal
- - chaos-distrib-x86_64-Linux-llvm/bin/TestFramework
+ - chaos-distrib-x86_64-Linux/bin/TestFramework --gtest_repeat=10 --gtest_break_on_failure
only:
- tags
- web
+ retry: 2
exp_llvm_scan_coverity:
tags:
@@ -313,11 +431,12 @@ exp_llvm_scan_coverity:
- /builds/chaos-lnf-control/chaosframework/config/CACHE/
only:
- schedules
+ retry: 2
pages:
stage: publish
dependencies:
- - build_framework_x86_64_test
+ - build_ubuntu_14_04_gcc48_test
script:
- mv coverage_html public
- ls -la
diff --git a/.vscode/launch.json b/.vscode/launch.json
index bd7744f80aa88d4bab15a0177932cd18674ffca7..0368f4a0383424de2dd01eaf49f4371a85730463 100644
--- a/.vscode/launch.json
+++ b/.vscode/launch.json
@@ -11,7 +11,7 @@
"osx": {
"program": "${workspaceFolder}/build/build-x86_64-Darwin/TestFramework",
},
- "args": ["--gtest_filter=PluginTest.*"],
+ "args": ["--gtest_filter=ExternalUnitTest.Echo"],
"stopAtEntry": false,
"cwd": "${workspaceFolder}",
"environment": [],
diff --git a/CHAOSFramework.xcodeproj/project.pbxproj b/CHAOSFramework.xcodeproj/project.pbxproj
index 631a5a1393ac5b748139b1f2beee18f8f90c1756..cda96f500c3ac8d8164f1240a91882e3a26a0962 100644
--- a/CHAOSFramework.xcodeproj/project.pbxproj
+++ b/CHAOSFramework.xcodeproj/project.pbxproj
@@ -7038,7 +7038,8 @@
buildSettings = {
CLANG_CXX_LANGUAGE_STANDARD = "gnu++0x";
CLANG_CXX_LIBRARY = "libc++";
- GCC_C_LANGUAGE_STANDARD = gnu99;
+ COMPILER_INDEX_STORE_ENABLE = NO;
+ GCC_C_LANGUAGE_STANDARD = gnu11;
GCC_PREPROCESSOR_DEFINITIONS = "";
HEADER_SEARCH_PATHS = (
.,
@@ -7715,7 +7716,8 @@
buildSettings = {
CLANG_CXX_LANGUAGE_STANDARD = "gnu++0x";
CLANG_CXX_LIBRARY = "libc++";
- GCC_C_LANGUAGE_STANDARD = gnu99;
+ COMPILER_INDEX_STORE_ENABLE = NO;
+ GCC_C_LANGUAGE_STANDARD = gnu11;
GCC_PREPROCESSOR_DEFINITIONS = "";
HEADER_SEARCH_PATHS = (
.,
@@ -7731,7 +7733,8 @@
buildSettings = {
CLANG_CXX_LANGUAGE_STANDARD = "gnu++0x";
CLANG_CXX_LIBRARY = "libc++";
- GCC_C_LANGUAGE_STANDARD = gnu99;
+ COMPILER_INDEX_STORE_ENABLE = NO;
+ GCC_C_LANGUAGE_STANDARD = gnu11;
GCC_PREPROCESSOR_DEFINITIONS = "";
HEADER_SEARCH_PATHS = (
.,
diff --git a/CHAOSFrameworkTests/CHAOSFrameworkTests.xcodeproj/project.pbxproj b/CHAOSFrameworkTests/CHAOSFrameworkTests.xcodeproj/project.pbxproj
index 033a71c188a1f6b07c041fc290ada7d2d6f0f665..2c35e07a5e1571d32c83131e9f22127758224ca1 100644
--- a/CHAOSFrameworkTests/CHAOSFrameworkTests.xcodeproj/project.pbxproj
+++ b/CHAOSFrameworkTests/CHAOSFrameworkTests.xcodeproj/project.pbxproj
@@ -81,6 +81,7 @@
325A84CC2108630C0036C1CA /* Base64Test.cpp */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.cpp.cpp; path = Base64Test.cpp; sourceTree = "<group>"; };
326274AE20B421C800BB0C91 /* DirectIOSystemServerChannelTest.cpp */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.cpp.cpp; path = DirectIOSystemServerChannelTest.cpp; sourceTree = "<group>"; };
32639FE81F6AAD990049089C /* main.cpp */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.cpp.cpp; path = main.cpp; sourceTree = "<group>"; };
+ 326C4E1E212AAEEF00DBCD8F /* TestTimer */ = {isa = PBXFileReference; explicitFileType = "compiled.mach-o.executable"; includeInIndex = 0; path = TestTimer; sourceTree = BUILT_PRODUCTS_DIR; };
3270978C209B440E00599038 /* ScriptClingTest.h */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.c.h; path = ScriptClingTest.h; sourceTree = "<group>"; };
3277A2302097586400FD7307 /* ScriptClingTest.cpp */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.cpp.cpp; path = ScriptClingTest.cpp; sourceTree = "<group>"; };
327A28691D71D72700DFDD2C /* TestCommandExecutor.cpp */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.cpp.cpp; path = TestCommandExecutor.cpp; sourceTree = "<group>"; };
@@ -122,7 +123,6 @@
32EBC21A1E92507B00785AAC /* main.cpp */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.cpp.cpp; path = main.cpp; sourceTree = "<group>"; };
32EBC2201E9250F900785AAC /* TimerTask.cpp */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.cpp.cpp; path = TimerTask.cpp; sourceTree = "<group>"; };
32EBC2211E9250F900785AAC /* TimerTask.h */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.c.h; path = TimerTask.h; sourceTree = "<group>"; };
- 32F842552010B315004D582D /* TestTimer */ = {isa = PBXFileReference; explicitFileType = "compiled.mach-o.executable"; includeInIndex = 0; name = TestTimer; path = /Users/bisegni/sources/chaos/chaos_development/chaosframework/CHAOSFrameworkTests/../usr/local/bin/TestTimer; sourceTree = "<absolute>"; };
32F842582010B315004D582D /* FrameworkTest */ = {isa = PBXFileReference; explicitFileType = "compiled.mach-o.executable"; includeInIndex = 0; path = FrameworkTest; sourceTree = BUILT_PRODUCTS_DIR; };
32F8425C20162B54004D582D /* FutureHelperTest.h */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.c.h; path = FutureHelperTest.h; sourceTree = "<group>"; };
32F8425D20162B54004D582D /* FutureHelperTest.cpp */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.cpp.cpp; path = FutureHelperTest.cpp; sourceTree = "<group>"; };
@@ -267,6 +267,7 @@
32639FE71F6AAD990049089C /* FrameworkTest */,
32F842582010B315004D582D /* FrameworkTest */,
3296E0752119E9100021C16B /* ExamplePlugin.chaos_extension */,
+ 326C4E1E212AAEEF00DBCD8F /* TestTimer */,
);
sourceTree = "<group>";
};
@@ -411,7 +412,7 @@
);
name = TestTimer;
productName = TestTimer;
- productReference = 32F842552010B315004D582D /* TestTimer */;
+ productReference = 326C4E1E212AAEEF00DBCD8F /* TestTimer */;
productType = "com.apple.product-type.tool";
};
/* End PBXNativeTarget section */
@@ -664,6 +665,7 @@
CLANG_WARN_UNREACHABLE_CODE = YES;
CLANG_WARN__DUPLICATE_METHOD_MATCH = YES;
CODE_SIGN_IDENTITY = "";
+ COMPILER_INDEX_STORE_ENABLE = NO;
CONFIGURATION_BUILD_DIR = "$(SRCROOT)/../usr/local/bin";
COPY_PHASE_STRIP = NO;
DEBUG_INFORMATION_FORMAT = dwarf;
@@ -717,6 +719,7 @@
CLANG_WARN_UNREACHABLE_CODE = YES;
CLANG_WARN__DUPLICATE_METHOD_MATCH = YES;
CODE_SIGN_IDENTITY = "";
+ COMPILER_INDEX_STORE_ENABLE = NO;
CONFIGURATION_BUILD_DIR = "$(SRCROOT)/../usr/local/bin";
COPY_PHASE_STRIP = NO;
DEBUG_INFORMATION_FORMAT = "dwarf-with-dsym";
diff --git a/CHAOSFrameworkTests/FrameworkTest/CMakeLists.txt b/CHAOSFrameworkTests/FrameworkTest/CMakeLists.txt
index 3d8233a6f44106e18ba870d3f7936d5d7af9f146..569d595e5fe789552be702a8727659b3f28d8a8b 100644
--- a/CHAOSFrameworkTests/FrameworkTest/CMakeLists.txt
+++ b/CHAOSFrameworkTests/FrameworkTest/CMakeLists.txt
@@ -1,6 +1,6 @@
cmake_minimum_required(VERSION 2.8)
project(TestFramework)
-
+#find_package(GTest REQUIRED)
IF( CHAOS_SANITIZER MATCHES ${PROJECT_NAME} )
MESG("ENABLING SANITIZER FOR PROJECT ${PROJECT_NAME}")
set (CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -g -fsanitize=address")
@@ -59,8 +59,7 @@ if(UNIX AND NOT APPLE)
LINK_DIRECTORIES(${PROJECT_SOURCE_DIR}/../../usr/local/lib/x86_64-linux-gnu)
endif()
ADD_EXECUTABLE(TestFramework ${TestFrameworkSource})
-
-
+#gtest_discover_tests(TestFramework)
add_dependencies(TestFramework ExamplePlugin)
IF(BUILD_FORCE_STATIC)
diff --git a/CHAOSFrameworkTests/FrameworkTest/direct_io/DirectIODeviceServerChannelTest.cpp b/CHAOSFrameworkTests/FrameworkTest/direct_io/DirectIODeviceServerChannelTest.cpp
index c08dc620d3e89ebaf39c28240cb27403028df6a4..3f2bf6354ad8b9b4d8b13e65cc44939940f32353 100644
--- a/CHAOSFrameworkTests/FrameworkTest/direct_io/DirectIODeviceServerChannelTest.cpp
+++ b/CHAOSFrameworkTests/FrameworkTest/direct_io/DirectIODeviceServerChannelTest.cpp
@@ -215,7 +215,7 @@ TEST_F(DirectIOChannelTest, DeviceChannelTest) {
client_channel = (DirectIODeviceClientChannel*)connection->getNewChannelInstance("DirectIODeviceClientChannel");
ASSERT_TRUE(client_channel);
//consumePutEvent
- for(int idx=0; idx < 1000; idx++)
+ for(int idx=0; idx < 100; idx++)
{
CDWUniquePtr push_data(new CDataWrapper());
push_data->addInt32Value("key", chaos::DataServiceNodeDefinitionType::DSStorageTypeLive);
@@ -229,7 +229,7 @@ TEST_F(DirectIOChannelTest, DeviceChannelTest) {
ASSERT_FALSE(client_channel->storeAndCacheDataOutputChannel("key", (void*)ser_2->getBufferPtr(), (uint32_t)ser_2->getBufferLen(), chaos::DataServiceNodeDefinitionType::DSStorageTypeLive));
}
- for(int idx=0; idx < 1000; idx++)
+ for(int idx=0; idx < 100; idx++)
{
CDWUniquePtr push_data(new CDataWrapper());
push_data->addInt32Value("key", chaos::DataServiceNodeDefinitionType::DSStorageTypeHistory);
@@ -243,7 +243,7 @@ TEST_F(DirectIOChannelTest, DeviceChannelTest) {
ASSERT_FALSE(client_channel->storeAndCacheDataOutputChannel("key", (void*)ser_2->getBufferPtr(), (uint32_t)ser_2->getBufferLen(), chaos::DataServiceNodeDefinitionType::DSStorageTypeHistory));
}
- for(int idx=0; idx < 1000; idx++)
+ for(int idx=0; idx < 100; idx++)
{
CDWUniquePtr push_data(new CDataWrapper());
push_data->addInt32Value("key", chaos::DataServiceNodeDefinitionType::DSStorageTypeLiveHistory);
@@ -257,7 +257,7 @@ TEST_F(DirectIOChannelTest, DeviceChannelTest) {
ASSERT_FALSE(client_channel->storeAndCacheDataOutputChannel("key", (void*)ser_2->getBufferPtr(), (uint32_t)ser_2->getBufferLen(), chaos::DataServiceNodeDefinitionType::DSStorageTypeLiveHistory));
}
- for(int idx=0; idx < 1000; idx++)
+ for(int idx=0; idx < 100; idx++)
{
CDWUniquePtr push_data(new CDataWrapper());
push_data->addInt32Value("key", chaos::DataServiceNodeDefinitionType::DSStorageTypeLive);
@@ -270,7 +270,7 @@ TEST_F(DirectIOChannelTest, DeviceChannelTest) {
ASSERT_TRUE(client_channel->storeAndCacheHealthData("key", (void*)ser_1->getBufferPtr(), (uint32_t)ser_1->getBufferLen(), chaos::DataServiceNodeDefinitionType::DSStorageTypeLive));
ASSERT_FALSE(client_channel->storeAndCacheHealthData("key", (void*)ser_2->getBufferPtr(), (uint32_t)ser_2->getBufferLen(), chaos::DataServiceNodeDefinitionType::DSStorageTypeLive));
}
- for(int idx=0; idx < 1000; idx++)
+ for(int idx=0; idx < 100; idx++)
{
CDWUniquePtr push_data(new CDataWrapper());
push_data->addInt32Value("key", chaos::DataServiceNodeDefinitionType::DSStorageTypeHistory);
@@ -284,7 +284,7 @@ TEST_F(DirectIOChannelTest, DeviceChannelTest) {
ASSERT_FALSE(client_channel->storeAndCacheHealthData("key", (void*)ser_2->getBufferPtr(), (uint32_t)ser_2->getBufferLen(), chaos::DataServiceNodeDefinitionType::DSStorageTypeHistory));
}
- for(int idx=0; idx < 1000; idx++)
+ for(int idx=0; idx < 100; idx++)
{
CDWUniquePtr push_data(new CDataWrapper());
push_data->addInt32Value("key", chaos::DataServiceNodeDefinitionType::DSStorageTypeLiveHistory);
@@ -298,7 +298,7 @@ TEST_F(DirectIOChannelTest, DeviceChannelTest) {
ASSERT_FALSE(client_channel->storeAndCacheHealthData("key", (void*)ser_2->getBufferPtr(), (uint32_t)ser_2->getBufferLen(), chaos::DataServiceNodeDefinitionType::DSStorageTypeLiveHistory));
}
- for(int idx=0; idx < 1000; idx++)
+ for(int idx=0; idx < 100; idx++)
{
get_counter = 0;
consumeGetEvent_counter = 0;
@@ -319,7 +319,7 @@ TEST_F(DirectIOChannelTest, DeviceChannelTest) {
ASSERT_STREQ(result_cdw->getStringValue("key").c_str(), "string");
free(result);result=NULL;
}
- for(int idx=0; idx < 1000; idx++)
+ for(int idx=0; idx < 100; idx++)
{
consumeGetEventMulti_counter = 0;
ChaosStringVector keys;
@@ -339,12 +339,12 @@ TEST_F(DirectIOChannelTest, DeviceChannelTest) {
ASSERT_STREQ(results[idx]->getStringValue("key").c_str(), keys[idx].c_str());
}
}
- for(int idx=0; idx < 1000; idx++)
+ for(int idx=0; idx < 100; idx++)
{
ASSERT_FALSE(client_channel->deleteDataCloud("search", std::numeric_limits<uint32_t>::min(), std::numeric_limits<uint32_t>::max()));
ASSERT_FALSE(client_channel->deleteDataCloud("search", std::numeric_limits<uint32_t>::min(), std::numeric_limits<uint32_t>::max()));
}
- for(int idx=0; idx < 1000; idx++)
+ for(int idx=0; idx < 100; idx++)
{
QueryResultPage results;
SearchSequence sseq = {1,2};
diff --git a/CHAOSFrameworkTests/FrameworkTest/external_unit/ExternalUnitMulticlientTest.cpp b/CHAOSFrameworkTests/FrameworkTest/external_unit/ExternalUnitMulticlientTest.cpp
index e49e358a1f6265a9e2acd11b47c750ba0b5feb1a..27e850b25b7c3c7c9b726c238d7c296023bf7cca 100644
--- a/CHAOSFrameworkTests/FrameworkTest/external_unit/ExternalUnitMulticlientTest.cpp
+++ b/CHAOSFrameworkTests/FrameworkTest/external_unit/ExternalUnitMulticlientTest.cpp
@@ -95,7 +95,7 @@ client_id(_client_id){
TestClient::~TestClient(){}
void TestClient::handleNewConnection(const std::string& connection_identifier){current_connection = connection_identifier; connection_event_counter++;}
-void TestClient::handleDisconnection(const std::string& connection_identifier){disconnection_event_counter++;}
+void TestClient::handleDisconnection(const std::string& connection_identifier){current_connection.clear(); disconnection_event_counter++;}
int TestClient::handleReceivedeMessage(const std::string& connection_identifier,
chaos::common::data::CDWUniquePtr message) {
if(message->hasKey(ECHO_TEST_KEY) &&
@@ -151,6 +151,7 @@ void clientRunner(ChaosSharedPtr<TestClient> client) {
}
TEST_F(ExternalUnitMulticlientTest, MultithreadingTest) {
+ ended = 0;
boost::thread_group tg;
for(int idx = 0; idx < NUMBER_OF_CLIENT; idx++) {
tg.add_thread(new boost::thread(clientRunner, ChaosSharedPtr<TestClient>(new TestClient(idx))));
diff --git a/CHAOSFrameworkTests/FrameworkTest/external_unit/ExternalUnitTest.cpp b/CHAOSFrameworkTests/FrameworkTest/external_unit/ExternalUnitTest.cpp
index a5f114cbb7b6f363dc6e40e736e2993b8129bc53..786e1c0e6c15941ddc17cbd60bd04c2235bec30d 100644
--- a/CHAOSFrameworkTests/FrameworkTest/external_unit/ExternalUnitTest.cpp
+++ b/CHAOSFrameworkTests/FrameworkTest/external_unit/ExternalUnitTest.cpp
@@ -92,6 +92,34 @@ TEST_F(ExternalUnitTest, InitDeinitCicle) {
ASSERT_NO_THROW(ExternalUnitManager::getInstance()->init(NULL));
}
+TEST_F(ExternalUnitTest, WrongEndpoint) {
+ int retry = 0;
+
+ ExternalUnitManager::getInstance()->initilizeConnection(*this,
+ "http",
+ "application/bson-json",
+ "ws://localhost:8080/badendpoint");
+ while(ExternalUnitClientEndpoint::isOnline() == false){
+ ASSERT_LE(retry++, 1000);
+ usleep(500000);
+ }
+ while(ExternalUnitClientEndpoint::getAcceptedState() != -1){
+ ASSERT_LE(retry++, 1000);
+ usleep(500000);
+ }
+
+ while(ExternalUnitClientEndpoint::isOnline() == true){
+ ASSERT_LE(retry++, 1000);
+ usleep(500000);
+ }
+
+ ASSERT_EQ(ExternalUnitClientEndpoint::isOnline(), false);
+ ASSERT_EQ(ExternalUnitClientEndpoint::getAcceptedState(), -1);
+ ExternalUnitManager::getInstance()->releaseConnection(*this,
+ "http");
+
+}
+
TEST_F(ExternalUnitTest, Echo) {
int retry = 0;
CDWUniquePtr message(new CDataWrapper());
@@ -103,7 +131,7 @@ TEST_F(ExternalUnitTest, Echo) {
"ws://localhost:8080/echo");
while(ExternalUnitClientEndpoint::isOnline() == false ||
ExternalUnitClientEndpoint::getAcceptedState() != 1) {
- ASSERT_LE(retry++, 10);
+ ASSERT_LE(retry++, 1000);
usleep(500000);
}
ASSERT_EQ(ExternalUnitClientEndpoint::isOnline(), true);
diff --git a/CHAOSFrameworkTests/FrameworkTest/main.cpp b/CHAOSFrameworkTests/FrameworkTest/main.cpp
index 08f81d4480a8e5c5277d3af0856b91012451b5ec..d484560e113a3f9c95cd61cc5a91b703b497cc97 100644
--- a/CHAOSFrameworkTests/FrameworkTest/main.cpp
+++ b/CHAOSFrameworkTests/FrameworkTest/main.cpp
@@ -25,9 +25,9 @@
#include <csignal>
int main(int argc, char ** argv) {
- char const * dummy_option[1] = {"--direct-io-client-kv-param=ZMQ_RCVTIMEO:600000"};
+ char const * dummy_option[2] = {"--log-on-console","--direct-io-client-kv-param=ZMQ_RCVTIMEO:600000"};
chaos::GlobalConfiguration::getInstance()->preParseStartupParameters();
- chaos::GlobalConfiguration::getInstance()->parseStartupParameters(1, dummy_option);
+ chaos::GlobalConfiguration::getInstance()->parseStartupParameters(2, dummy_option);
chaos::common::log::LogManager::getInstance()->init();
::testing::InitGoogleTest(&argc, argv);
diff --git a/CHAOSFrameworkTests/FrameworkTest/utility/FutureHelperTest.cpp b/CHAOSFrameworkTests/FrameworkTest/utility/FutureHelperTest.cpp
index c2efe604de61d9d59f057d09102898517dc240de..92507f8e104acde5a86c669b849c33086c066a1d 100644
--- a/CHAOSFrameworkTests/FrameworkTest/utility/FutureHelperTest.cpp
+++ b/CHAOSFrameworkTests/FrameworkTest/utility/FutureHelperTest.cpp
@@ -28,7 +28,7 @@ uint32_t future_to_counter = 0;
uint32_t promises_counter = 0;
using namespace chaos::common::data;
-#define NUMBER_OF_TEST 1000000
+#define NUMBER_OF_TEST 1000
using namespace chaos::common::utility;
using namespace chaos::common::async_central;
@@ -63,6 +63,10 @@ TEST(FutureHelperTests, Base) {
//boost::thread_group threads;
CDWComsumerPromise pq;
CDWComsumerFuture fq;
+ future_counter = 0;
+ future_excpt_counter = 0;
+ future_to_counter = 0;
+ promises_counter = 0;
ASSERT_NO_THROW(InizializableService::initImplementation(AsyncCentralManager::getInstance(), NULL, "AsyncCentralManager", __PRETTY_FUNCTION__));
MessageRequestDomainFutureHelperShrdPtr helper_test(new MessageRequestDomainFutureHelper(5000,30000));
MessageRequestDomainFutureHelper::Future new_shared_future;
diff --git a/CMakeLists.txt b/CMakeLists.txt
index 86f06098afc05859f914eb1ecc0bab74a661ef2e..eb27541447f21d967732d07b697dd8104d75bef1 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -4,6 +4,8 @@ include(config/CMakeChaos.txt)
project(chaosframework)
include(ExternalProject)
#include(CTest)
+#enable_testing()
+
#find git
find_package(Git REQUIRED)
@@ -94,7 +96,7 @@ IF (USE_GPT)
PREFIX "${CMAKE_BINARY_DIR}/ext_dep/gpt-prefix"
SOURCE_DIR "${CMAKE_BINARY_DIR}/ext_dep/gpt-src"
BINARY_DIR "${CMAKE_BINARY_DIR}/ext_dep/gpt-src"
- CONFIGURE_COMMAND ${CMAKE_CURRENT_SOURCE_DIR}/ext_dep/gpt-src/./autogen.sh COMMAND ${CMAKE_CURRENT_SOURCE_DIR}/ext_dep/gpt-src/./configure --prefix=${CMAKE_INSTALL_PREFIX}
+ CONFIGURE_COMMAND ${CMAKE_BINARY_DIR}/ext_dep/gpt-src/./autogen.sh COMMAND ${CMAKE_BINARY_DIR}/ext_dep/gpt-src/./configure --prefix=${CMAKE_INSTALL_PREFIX}
LOG_DOWNLOAD ON
LOG_CONFIGURE ON
LOG_BUILD ON)
diff --git a/chaos/common/ChaosCommon.h b/chaos/common/ChaosCommon.h
index b2ed1f56a8d325d7f8c4b62968c0394fb6557a15..0aa99a8b6f840dcf6bbfafa6fcb6316d19c86709 100644
--- a/chaos/common/ChaosCommon.h
+++ b/chaos/common/ChaosCommon.h
@@ -88,7 +88,7 @@ namespace chaos {
#endif
}
-#include <csignal>
+//#include <csignal>
#ifndef CHAOS_NO_BACKTRACE
class SignalHandling {
public:
@@ -145,19 +145,23 @@ namespace chaos {
bool loaded() const { return _loaded; }
static void handleSignal(int, siginfo_t* info, void* _ctx) {
- ucontext_t *uctx = (ucontext_t*) _ctx;
backward::StackTrace st;
void* error_addr = 0;
#if defined(REG_RIP)// x86_64
+ ucontext_t *uctx = (ucontext_t*) _ctx;
error_addr = reinterpret_cast<void*>(uctx->uc_mcontext.gregs[REG_RIP]);
#elif defined(REG_EIP) // x86_32
+ ucontext_t *uctx = (ucontext_t*) _ctx;
error_addr = reinterpret_cast<void*>(uctx->uc_mcontext.gregs[REG_EIP]);
#elif defined(__arm__)
+ ucontext_t *uctx = (ucontext_t*) _ctx;
error_addr = reinterpret_cast<void*>(uctx->uc_mcontext.arm_pc);
#elif defined(__aarch64__)
+ ucontext_t *uctx = (ucontext_t*) _ctx;
error_addr = reinterpret_cast<void*>(uctx->uc_mcontext.pc);
#elif defined(__ppc__) || defined(__powerpc) || defined(__powerpc__) || defined(__POWERPC__)
+ ucontext_t *uctx = (ucontext_t*) _ctx;
error_addr = reinterpret_cast<void*>(uctx->uc_mcontext.regs->nip);
#else
#endif
diff --git a/chaos/common/configuration/GlobalConfiguration.cpp b/chaos/common/configuration/GlobalConfiguration.cpp
index f1eb0000d63198e1de8f5511df6b985abb877807..2428ea215fb5bf9bf95fb8f0c71b37fa584d7957 100644
--- a/chaos/common/configuration/GlobalConfiguration.cpp
+++ b/chaos/common/configuration/GlobalConfiguration.cpp
@@ -234,7 +234,7 @@ void GlobalConfiguration::checkDefaultOption() throw (CException) {
CHECK_AND_DEFINE_OPTION(string, logSyslogSrv, InitOption::OPT_LOG_SYSLOG_SERVER);
configuration->addStringValue(InitOption::OPT_LOG_SYSLOG_SERVER, logSyslogSrv);
- CHECK_AND_DEFINE_OPTION(uint32_t, logSyslogSrvPort, InitOption::OPT_LOG_SYSLOG_SERVER_PORT);
+ CHECK_AND_DEFINE_OPTION_WITH_DEFAULT(uint32_t, logSyslogSrvPort, InitOption::OPT_LOG_SYSLOG_SERVER_PORT, 0);
configuration->addInt32Value(InitOption::OPT_LOG_SYSLOG_SERVER_PORT, logSyslogSrvPort);
CHECK_AND_DEFINE_BOOL_ZERO_TOKEN_OPTION(logOnFile, InitOption::OPT_LOG_ON_FILE);
diff --git a/chaos/common/configuration/GlobalConfiguration.h b/chaos/common/configuration/GlobalConfiguration.h
index c401efe82014dc7f6b961858dceceb0edfe8f283..6a75440a9071ee202aa9ed5e850a215b70740779 100644
--- a/chaos/common/configuration/GlobalConfiguration.h
+++ b/chaos/common/configuration/GlobalConfiguration.h
@@ -49,12 +49,10 @@ namespace chaos {
using namespace std;
#define CHECK_AND_DEFINE_OPTION_WITH_DEFAULT(t,x,y,z)\
-t x;\
+t x = z;\
if(hasOption(y)){\
x = getOption<t>(y);\
-}else{\
-x = z;\
-}\
+}
#define CHECK_OPTION_WITH_DEFAULT(t,x,y,z)\
if(hasOption(y)){\
diff --git a/chaos/common/cronus_manager/CronJob.h b/chaos/common/cronus_manager/CronJob.h
index 548261562d641fa7f9e95d60081ad46f0a7602e5..7563739d2a23d2ccae858f279e731910a847ed9b 100644
--- a/chaos/common/cronus_manager/CronJob.h
+++ b/chaos/common/cronus_manager/CronJob.h
@@ -74,7 +74,7 @@ namespace chaos {
void log(const std::string& log_message);
public:
CronJob(chaos::common::data::CDataWrapper *job_parameter);
- ~CronJob();
+ virtual ~CronJob();
};
}
}
diff --git a/chaos/common/data/cache/DataCache.cpp b/chaos/common/data/cache/DataCache.cpp
index e0ba3b4b69f99da1353b44a78d5ce72fe0d5980f..2e9ecd1265cab1b833282bca1ee72e849824b1f0 100644
--- a/chaos/common/data/cache/DataCache.cpp
+++ b/chaos/common/data/cache/DataCache.cpp
@@ -26,8 +26,6 @@
using namespace chaos::common::data::cache;
-const int initial_pool_size = 64;
-
DataCache::DataCache(){
hashpower = 16;
primary_hashtable = NULL;
@@ -128,7 +126,7 @@ int DataCache::storeItem(const char *key, const void *buffer, uint32_t bufferLen
//! delete item
int DataCache::deleteItem(const char *key) {
- boost::unique_lock<boost::shared_mutex>(mc_mutex);
+ boost::unique_lock<boost::shared_mutex> ul(mc_mutex);
item *it = do_item_get(key, strlen(key));
if(!it) {
return -1;
@@ -321,7 +319,7 @@ int DataCache::start_assoc_maintenance_thread() {
}
void DataCache::stop_assoc_maintenance_thread() {
- boost::unique_lock<boost::mutex>(mc_mutex);
+ boost::unique_lock<boost::shared_mutex> ul(mc_mutex);
do_run_maintenance_thread = 0;
pthread_cond_signal(&maintenance_cond);
@@ -548,4 +546,4 @@ int DataCache::do_item_replace(item *it, item *new_it) {
return do_item_link(new_it);
}
-#pragma GCC visibility pop
\ No newline at end of file
+#pragma GCC visibility pop
diff --git a/chaos/common/data/entity/Entity.cpp b/chaos/common/data/entity/Entity.cpp
index f38f159ad875b58f2f4e9c15ba2eed5ed3623a42..7411b6648b9a75d6a3c8501c6ca975d0a0255801 100644
--- a/chaos/common/data/entity/Entity.cpp
+++ b/chaos/common/data/entity/Entity.cpp
@@ -100,7 +100,6 @@ int32_t Entity::getAllProperty(ptr_vector<chaos::edb::KeyIdAndValue>& propertys)
}
int32_t Entity::getPropertyByKeyID(uint32_t keyID, ptr_vector<chaos::edb::KeyIdAndValue>& propertys) {
- int32_t error = 0;
vector<uint32_t> keys;
keys.push_back(keyID);
return database->searchPropertyForEntity(entityID, keys, propertys);
diff --git a/chaos/common/direct_io/DirectIOClientConnectionMetricCollector.cpp b/chaos/common/direct_io/DirectIOClientConnectionMetricCollector.cpp
index ecb3624bb98b31b9a22414438c5a6dae2a2c6d7c..2b01a21595abace799f31137b9c917b98f683914 100644
--- a/chaos/common/direct_io/DirectIOClientConnectionMetricCollector.cpp
+++ b/chaos/common/direct_io/DirectIOClientConnectionMetricCollector.cpp
@@ -25,8 +25,6 @@
using namespace chaos::common::direct_io;
-static const char * const METRIC_KEY_ENDPOINT_ALIVE = "ndpoint_alive";
-
#define DIOCCMC_INFO INFO_LOG(DirectIOClientConnectionMetricCollector)
#define DIOCCMC_DBG_ DBG_LOG(DirectIOClientConnectionMetricCollector)
#define DIOCCMC_ERR_ ERR_LOG(DirectIOClientConnectionMetricCollector)
diff --git a/chaos/common/external_unit/ExternalEchoEndpoint.cpp b/chaos/common/external_unit/ExternalEchoEndpoint.cpp
index e94864ade840a5e9f7b91f9a4cad773afc2c1b70..11898e349c80fcffdbc9d06412eceabcc255748e 100644
--- a/chaos/common/external_unit/ExternalEchoEndpoint.cpp
+++ b/chaos/common/external_unit/ExternalEchoEndpoint.cpp
@@ -44,6 +44,7 @@ void ExternalEchoEndpoint::handleDisconnection(const std::string& connection_ide
int ExternalEchoEndpoint::handleReceivedeMessage(const std::string& connection_identifier,
chaos::common::data::CDWUniquePtr message) {
+ message_counter++;
INFO << CHAOS_FORMAT("Received connection from %1% with data '%2%'", %connection_identifier%message->getJSONString());
message->addStringValue("ExternalEchoEndpoint", "echo answer");
sendMessage(connection_identifier, ChaosMoveOperator(message));
diff --git a/chaos/common/external_unit/ExternalEchoEndpoint.h b/chaos/common/external_unit/ExternalEchoEndpoint.h
index 1f743baa3928270dadbed3aba3ce0006badff479..2959617bd56ab34b0f2ba982410f13e68fc7add9 100644
--- a/chaos/common/external_unit/ExternalEchoEndpoint.h
+++ b/chaos/common/external_unit/ExternalEchoEndpoint.h
@@ -22,6 +22,7 @@
#ifndef __CHAOSFramework_E6EB27EC_C56A_4A03_BF1C_03373F140079_ExternalEchoEndpoint_h
#define __CHAOSFramework_E6EB27EC_C56A_4A03_BF1C_03373F140079_ExternalEchoEndpoint_h
+#include <chaos/common/chaos_types.h>
#include <chaos/common/external_unit/ExternalUnitServerEndpoint.h>
namespace chaos{
@@ -30,7 +31,7 @@ namespace chaos{
class ExternalEchoEndpoint:
public ExternalUnitServerEndpoint {
- unsigned int message_counter;
+ ChaosAtomic<uint32_t> message_counter;
protected:
//!inherited method by @ExternalUnitEndpoint
void handleNewConnection(const std::string& connection_identifier);
diff --git a/chaos/common/external_unit/http_adapter/HTTPBaseAdapter.h b/chaos/common/external_unit/http_adapter/HTTPBaseAdapter.h
index e93f9884d5999494ef03b47909bc8c7c5ba4768f..a6a0ef3b01ed57de1c65f8d0b854d5887c0960ed 100644
--- a/chaos/common/external_unit/http_adapter/HTTPBaseAdapter.h
+++ b/chaos/common/external_unit/http_adapter/HTTPBaseAdapter.h
@@ -38,60 +38,53 @@ namespace chaos {
namespace external_unit {
namespace http_adapter {
- struct ServerWorkRequest {
- WorkRequestType r_type;
- mg_connection *nc;
- std::string uri;
- //!serialization type issuead on the beginning of the http request
- std::string s_type;
- ChaosUniquePtr<chaos::common::data::CDataBuffer> buffer;
-
- ServerWorkRequest():
- r_type(WorkRequestTypeUnspecified),
- nc(NULL),
- uri(),
- s_type(),
- buffer(){}
-
- ServerWorkRequest(const char *ptr,
- uint32_t size):
- r_type(WorkRequestTypeUnspecified),
- nc(NULL),
- uri(),
- s_type(),
- buffer(new chaos::common::data::CDataBuffer(ptr, size)){}
+ /**
+ * @brief tag the conenction to be identified with the
+ * connection virtualization within the http adapter instance
+ */
+ template<typename T>
+ struct ConnectionMetadata {
+ std::string conn_uuid;
+ T *class_instance;
+ ConnectionMetadata(const std::string& _conn_uuid,
+ T *_class_instance):
+ conn_uuid(_conn_uuid),
+ class_instance(_class_instance){}
};
-
- typedef ChaosSharedPtr<ServerWorkRequest> ServerWorkRequestShrdPtr;
-
+
+ //!opcode list
+ typedef enum {
+ OpcodeInfoTypeSend,
+ OpcodeInfoTypeCloseConnection,
+ OpcodeInfoTypeCloseConnectionForEndpoint
+ } OpcodeInfoType;
+
+ struct Opcode {
+ //! context sensitive identifier associated to the operation
+ std::string identifier;
+ //! operation type
+ OpcodeInfoType op_type;
+ //!data to sent
+ chaos::common::data::CDBufferUniquePtr data;
+ //! send data opcode
+ EUCMessageOpcode data_opcode;
+ //some opcode need to notify the termination
+ chaos::WaitSemaphore wait_termination_semaphore;
+
+ Opcode(){}
+ virtual ~Opcode(){}
+ };
+
+ typedef ChaosSharedPtr<Opcode> OpcodeShrdPtr;
+ typedef std::queue<OpcodeShrdPtr> OpcodeShrdPtrQueue;
+ CHAOS_DEFINE_LOCKABLE_OBJECT(OpcodeShrdPtrQueue, LOpcodeShrdPtrQueue);
+
+ /*!
+ */
class HTTPBaseAdapter {
friend class ExternalUnitConnection;
protected:
- //structure for manage command executed in other thread
-
- //!opcode list
- typedef enum {
- OpcodeInfoTypeSend,
- OpcodeInfoTypeCloseConnection,
- OpcodeInfoTypeCloseConnectionForEndpoint
- } OpcodeInfoType;
-
- typedef struct Opcode {
- //! context sensitive identifier associated to the operation
- std::string identifier;
- //! operation type
- OpcodeInfoType op_type;
- //!data to sent
- chaos::common::data::CDBufferUniquePtr data;
- //! send data opcode
- EUCMessageOpcode data_opcode;
- //some opcode need to notify the termination
- chaos::WaitSemaphore wait_termination_semaphore;
- } Opcode;
-
- typedef ChaosSharedPtr<Opcode> OpcodeShrdPtr;
- typedef std::queue<OpcodeShrdPtr> OpcodeShrdPtrQueue;
- CHAOS_DEFINE_LOCKABLE_OBJECT(OpcodeShrdPtrQueue, LOpcodeShrdPtrQueue);
+ struct mg_mgr mgr;
//!operation posted during poll execution to send with the nex one
LOpcodeShrdPtrQueue post_evt_op_queue;
@@ -103,6 +96,20 @@ namespace chaos {
virtual int closeConnection(const std::string& connection_identifier) = 0;
+ //!find conneciton by uuid
+ template<typename T>
+ struct mg_connection* findConnection(const std::string& conenction_uuid) {
+ struct mg_connection *c = NULL;
+ for (c = mg_next(&mgr, NULL); c != NULL; c = mg_next(&mgr, c)) {
+ if(!c->user_data) continue;
+ ConnectionMetadata<T> *conn_metadata = static_cast<ConnectionMetadata<T> * >(c->user_data);
+ if(conn_metadata->conn_uuid.compare(conenction_uuid) == 0) {
+ break;
+ }
+ }
+ return c;
+ }
+
void sendHTTPJSONError(mg_connection *nc,
int status_code,
const int error_code,
@@ -115,6 +122,21 @@ namespace chaos {
mg_printf(nc, "%s", json_error.c_str());
}
+ OpcodeShrdPtr composeJSONErrorResposneOpcode(const std::string& connection_uuid,
+ const int error_code,
+ const std::string& error_message) {
+ chaos::common::data::CDataWrapper err_data_pack;
+ err_data_pack.addInt32Value("error_code", error_code);
+ err_data_pack.addStringValue("error_message", error_message);
+ const std::string json_error = err_data_pack.getCompliantJSONString();
+ OpcodeShrdPtr op(new Opcode());
+ op->identifier = connection_uuid;
+ op->op_type = OpcodeInfoTypeSend;
+ op->data = chaos::common::data::CDBufferUniquePtr(new chaos::common::data::CDataBuffer(json_error.c_str(), (uint32_t)json_error.size()));
+ op->data_opcode = EUCMessageOpcodeWhole;
+ return op;
+ }
+
void sendWSJSONError(mg_connection *nc,
const int error_code,
const std::string& error_message,
diff --git a/chaos/common/external_unit/http_adapter/HTTPClientAdapter.cpp b/chaos/common/external_unit/http_adapter/HTTPClientAdapter.cpp
index dc239a375ffc4215b01f50e848115f44d4e00dac..9f6ceba50909ce408c492151b63694a3c32a5e28 100644
--- a/chaos/common/external_unit/http_adapter/HTTPClientAdapter.cpp
+++ b/chaos/common/external_unit/http_adapter/HTTPClientAdapter.cpp
@@ -35,20 +35,17 @@ using namespace chaos::common::utility;
using namespace chaos::common::external_unit::http_adapter;
static const char *web_socket_option="Content-Type: application/bson-json\r\n";
-
HTTPClientAdapter::HTTPClientAdapter():
-run(false){
+run(false),
+message_broadcasted(0){
if(GlobalConfiguration::getInstance()->hasOption(chaos::InitOption::OPT_REST_POLL_TIME_US)){
rest_poll_time=GlobalConfiguration::getInstance()->getOption<uint32_t>(chaos::InitOption::OPT_REST_POLL_TIME_US);
} else {
- rest_poll_time=10;
+ rest_poll_time=1000;
}
-
}
-HTTPClientAdapter::~HTTPClientAdapter() {
-
-}
+HTTPClientAdapter::~HTTPClientAdapter() {}
void HTTPClientAdapter::init(void *init_data) throw (chaos::CException) {
mg_mgr_init(&mgr, NULL);
@@ -59,9 +56,7 @@ void HTTPClientAdapter::init(void *init_data) throw (chaos::CException) {
void HTTPClientAdapter::deinit() throw (chaos::CException) {
run = false;
DBG<<" HTTPClientAdapter DEINIT";
-
thread_poller->join();
-
mg_mgr_free(&mgr);
}
@@ -74,129 +69,134 @@ void HTTPClientAdapter::poller() {
usleep(rest_poll_time);
if(poll_counter++ % (rest_poll_time)*10000000){performReconnection();}
}
- //consume opcode queue
+
+ //check connection close opcode
{
- LOpcodeShrdPtrQueueReadLock wconnl = post_evt_op_queue.getReadLockObject();
+ LOpcodeShrdPtrQueueWriteLock wlm = post_evt_op_queue.getWriteLockObject();
while(post_evt_op_queue().empty() == false) {
OpcodeShrdPtr op = post_evt_op_queue().front();
post_evt_op_queue().pop();
- wconnl->unlock();
- switch(op->op_type) {
- case OpcodeInfoTypeSend:{
- LMapReconnectionInfoWriteLock wlm = map_connection.getWriteLockObject();
- MapReconnectionInfoIterator conn_it = map_connection().find(op->identifier);
- if(conn_it == map_connection().end()) {break;}
- if(conn_it->second->conn) {
- if(conn_it->second->ext_unit_conn->online == false) {break;};
- switch (op->data_opcode) {
- case EUCMessageOpcodeWhole:
- mg_send_websocket_frame(conn_it->second->conn, WEBSOCKET_OP_TEXT, op->data->getBuffer(), op->data->getBufferSize());
- break;
- case EUCPhaseStartFragment:
- mg_send_websocket_frame(conn_it->second->conn, WEBSOCKET_OP_TEXT|WEBSOCKET_DONT_FIN, op->data->getBuffer(), op->data->getBufferSize());
- break;
- case EUCPhaseContinueFragment:
- mg_send_websocket_frame(conn_it->second->conn, WEBSOCKET_OP_TEXT|WEBSOCKET_DONT_FIN, op->data->getBuffer(), op->data->getBufferSize());
- break;
- case EUCPhaseEndFragment:
- mg_send_websocket_frame(conn_it->second->conn, WEBSOCKET_OP_TEXT, op->data->getBuffer(), op->data->getBufferSize());
- break;
- }
- }
- break;
- }
- case OpcodeInfoTypeCloseConnection:{
- LMapReconnectionInfoWriteLock wlm = map_connection.getWriteLockObject();
- MapReconnectionInfoIterator conn_it = map_connection().find(op->identifier);
- if(conn_it == map_connection().end()) {break;};
- if(conn_it->second->conn) {
+ {
+ //close real connection
+ struct mg_connection *c = NULL;
+ for (c = mg_next(&mgr, NULL); c != NULL; c = mg_next(&mgr, c)) {
+ if(!c->user_data) continue;
+ ConnectionMetadata<HTTPClientAdapter> *conn_metadata = static_cast<ConnectionMetadata<HTTPClientAdapter> * >(c->user_data);
+ if(conn_metadata->conn_uuid.compare(op->identifier) == 0) {
DBG<<" HTTPClientAdapter Close Connection";
- conn_it->second->conn->flags |= MG_F_CLOSE_IMMEDIATELY;
- conn_it->second->conn->user_data = NULL;
+ c->flags |= MG_F_CLOSE_IMMEDIATELY;
+ delete(conn_metadata);
+ c->user_data = NULL;
+ break;
}
- //!remove from active connection map
+ }
+ //remove virtual one
+ LMapConnectionInfoWriteLock wlm = map_connection.getWriteLockObject();
+ MapConnectionInfoIterator conn_it = map_connection().find(op->identifier);
+ if(conn_it != map_connection().end()){
map_connection().erase(conn_it);
- break;
}
- default:{break;}
}
- wconnl->lock();
+ op->wait_termination_semaphore.unlock();
}
}
}
DBG<<" HTTPClientAdapter POLL EXIT";
-
INFO << "Leaving thread poller";
}
int HTTPClientAdapter::addNewConnectionForEndpoint(ExternalUnitClientEndpoint *endpoint,
const std::string& endpoint_url,
const std::string& serialization) {
- LMapReconnectionInfoWriteLock wlm = map_connection.getWriteLockObject();
+ LMapConnectionInfoWriteLock wlm = map_connection.getWriteLockObject();
ChaosUniquePtr<serialization::AbstractExternalSerialization> serializer = ExternalUnitManager::getInstance()->getNewSerializationInstanceForType(serialization);
if(!serializer.get()) {
return -1;
}
try{
- ChaosSharedPtr<ExternalUnitConnection> conn_ptr(new ExternalUnitConnection(this,
- endpoint,
- ChaosMoveOperator(serializer)));
+ ConnectionInfoShrdPtr ci = ConnectionInfoShrdPtr(new ConnectionInfo(endpoint_url));
+ ci->ext_unit_conn = ChaosSharedPtr<ExternalUnitConnection>(new ExternalUnitConnection(this,
+ endpoint,
+ ChaosMoveOperator(serializer)));
//!associate id to connection
- ConnectionInfoShrdPtr ci(new ConnectionInfo());
- map_connection().insert(MapReconnectionInfoPair(conn_ptr->connection_identifier, ci));
- ci->class_instance = this;
- ci->endpoint_url = endpoint_url;
- ci->ext_unit_conn = conn_ptr;
- ci->conn = mg_connect_ws(&mgr,
- HTTPClientAdapter::ev_handler,
- ci->endpoint_url.c_str(),
- "ChaosExternalUnit",
- web_socket_option);
- ci->conn->user_data = ci.get();
+ map_connection().insert(MapConnectionInfoPair(ci->ext_unit_conn->connection_identifier, ci));
+
+ mg_connection *conn = mg_connect_ws(&mgr,
+ HTTPClientAdapter::ev_handler,
+ ci->endpoint_url.c_str(),
+ "ChaosExternalUnit",
+ web_socket_option);
+ ci->ext_unit_conn->online = true;
+ conn->user_data = new ConnectionMetadata<HTTPClientAdapter>(ci->ext_unit_conn->connection_identifier, this);
} catch(chaos::CException& ex) {
return -2;
}
return 0;
}
+int HTTPClientAdapter::sendDataToConnection(const std::string& connection_identifier,
+ chaos::common::data::CDBufferUniquePtr data,
+ const EUCMessageOpcode opcode) {
+ LMapConnectionInfoReadLock wlm = map_connection.getReadLockObject();
+ MapConnectionInfoIterator conn_it = map_connection().find(connection_identifier);
+ if(conn_it == map_connection().end()) return 0;
+ OpcodeShrdPtr op(new Opcode());
+ op->identifier = connection_identifier;
+ op->op_type = OpcodeInfoTypeSend;
+ op->data = ChaosMoveOperator(data);
+ op->data_opcode = opcode;
+ ChaosWriteLock conn_wl(conn_it->second->smutex);
+ conn_it->second->opcode_queue.push(op);
+ return 0;
+}
+
int HTTPClientAdapter::removeConnectionsFromEndpoint(ExternalUnitClientEndpoint *target_endpoint) {
//remove endpoint from coon abstraction
- LMapReconnectionInfoWriteLock wlm = map_connection.getWriteLockObject();
- MapReconnectionInfoIterator it = map_connection().find(target_endpoint->getConnectionIdentifier());
- if(it == map_connection().end()) {return 0;}
- {
- LOpcodeShrdPtrQueueWriteLock wconnl = post_evt_op_queue.getWriteLockObject();
- OpcodeShrdPtr op(new Opcode());
- op->identifier = target_endpoint->getConnectionIdentifier();
- op->op_type = OpcodeInfoTypeCloseConnection;
- post_evt_op_queue().push(op);
- }
- //detach external unit cnnection abstraction
- it->second->ext_unit_conn.reset();
+ LOpcodeShrdPtrQueueWriteLock wlm = post_evt_op_queue.getWriteLockObject();
+ OpcodeShrdPtr op(new Opcode());
+ op->identifier = target_endpoint->getConnectionIdentifier();
+ op->op_type = OpcodeInfoTypeCloseConnection;
+ post_evt_op_queue().push(op);
+ wlm->unlock();
+ op->wait_termination_semaphore.wait();
+ return 0;
+}
+
+int HTTPClientAdapter::closeConnection(const std::string& connection_identifier) {
+ //remove endpoint from coon abstraction
+ LOpcodeShrdPtrQueueWriteLock wlm = post_evt_op_queue.getWriteLockObject();
+ OpcodeShrdPtr op(new Opcode());
+ op->identifier = connection_identifier;
+ op->op_type = OpcodeInfoTypeCloseConnection;
+ post_evt_op_queue().push(op);
+ wlm->unlock();
+ op->wait_termination_semaphore.wait();
return 0;
}
void HTTPClientAdapter::performReconnection() {
uint64_t cur_ts = TimingUtil::getTimeStamp();
//!reconnection
- LMapReconnectionInfoReadLock wlm = map_connection.getReadLockObject();
- for(MapReconnectionInfoIterator it = map_connection().begin(), end = map_connection().end();
+ LMapConnectionInfoReadLock wlm = map_connection.getReadLockObject();
+ for(MapConnectionInfoIterator it = map_connection().begin(), end = map_connection().end();
it != end;
it++) {
- if(it->second->conn == NULL &&
+ if(it->second->ext_unit_conn.get() &&
+ it->second->ext_unit_conn->online == false &&
cur_ts >= it->second->next_reconnection_retry_ts) {
//try to reconnect
- it->second->conn = mg_connect_ws(&mgr,
- HTTPClientAdapter::ev_handler,
- it->second->endpoint_url.c_str(),
- "ChaosExternalUnit",
- web_socket_option);
- if(it->second->conn) {
- it->second->conn->user_data = it->second.get();
+ mg_connection * conn = mg_connect_ws(&mgr,
+ HTTPClientAdapter::ev_handler,
+ it->second->endpoint_url.c_str(),
+ "ChaosExternalUnit",
+ web_socket_option);
+ if(conn) {
+ it->second->ext_unit_conn->online = true;
+ conn->user_data = new ConnectionMetadata<HTTPClientAdapter>(it->second->ext_unit_conn->connection_identifier, this);
} else {
//retry to reconnect
it->second->next_reconnection_retry_ts = TimingUtil::getTimestampWithDelay(5000, true);
@@ -218,39 +218,47 @@ void HTTPClientAdapter::checkAcceptResponse(struct websocket_message *wm,
void HTTPClientAdapter::ev_handler(struct mg_connection *conn,
int event,
void *event_data) {
- ConnectionInfo *ci = static_cast<ConnectionInfo*>(conn->user_data);
- if(ci == NULL) return;
+ if(!conn->user_data) return;
+ ConnectionMetadata<HTTPClientAdapter> *conn_metadata = static_cast<ConnectionMetadata<HTTPClientAdapter> * >(conn->user_data);
+
+ //get connection info
+ LMapConnectionInfoReadLock wlm = conn_metadata->class_instance->map_connection.getReadLockObject();
+ MapConnectionInfoIterator conn_it = conn_metadata->class_instance->map_connection().find(conn_metadata->conn_uuid);
+ if(conn_it == conn_metadata->class_instance->map_connection().end()) return;
+
+ ConnectionInfoShrdPtr conn_info = conn_it->second;
switch (event) {
case MG_EV_CONNECT: {
DBG<<" HTTP Client Connection event";
- ChaosWriteLock wl(ci->smux);
int status = *((int *) event_data);
- ci->ext_unit_conn->online = (status==0);
+ conn_info->ext_unit_conn->online = (status==0);
break;
}
case MG_EV_WEBSOCKET_FRAME: {
-
- CHAOS_ASSERT(ci->ext_unit_conn.get() != NULL);
int err = 0;
struct websocket_message *wm = (struct websocket_message *) event_data;
- if(ci->ext_unit_conn->accepted_state <= 0) {
+ if(conn_info->ext_unit_conn->accepted_state <= 0) {
bool is_accept_response = false;
int accept_result = -1;
//check accepted state
checkAcceptResponse(wm, is_accept_response, accept_result);
if(is_accept_response) {
- ci->ext_unit_conn->accepted_state = accept_result;
+ conn_info->ext_unit_conn->accepted_state = accept_result;
} else {
- ci->class_instance->sendWSJSONError(conn, -2, "Accept response is not well formed!");
+ conn_info->ext_unit_conn->accepted_state = accept_result;
+ std::string json_string((const char *)wm->data, wm->size);
+ DBG << json_string;
}
} else {
- //accepted connection ca received data
+ //accepted connection can received data
ChaosUniquePtr<CDataBuffer> buffer(new CDataBuffer((const char *)wm->data,
(uint32_t)wm->size));
- if((err = ci->class_instance->sendDataToEndpoint(*ci->ext_unit_conn,
- ChaosMoveOperator(buffer)))) {
+ if((err = conn_metadata->class_instance->sendDataToEndpoint(*conn_info->ext_unit_conn,
+ ChaosMoveOperator(buffer)))) {
//weh don't have found the sriealizer
- ci->class_instance->sendWSJSONError(conn, err, "Error sending data to endpoint");
+ ERR<< CHAOS_FORMAT("Error forwading data from connection uuid %1%", %conn_info->ext_unit_conn->connection_identifier);
+ } else {
+ conn_metadata->class_instance->message_broadcasted++;
}
}
break;
@@ -258,41 +266,47 @@ void HTTPClientAdapter::ev_handler(struct mg_connection *conn,
case MG_EV_CLOSE: {
//manage the reconnection
DBG<<" HTTP Client CLOSE event";
- LMapReconnectionInfoReadLock wlm = ci->class_instance->map_connection.getReadLockObject();
- if(ci->ext_unit_conn.get() == NULL) {break;}
- if(ci->class_instance->map_connection().count(ci->ext_unit_conn->connection_identifier) !=0) {
- //in this case concnretion info need to be put into reconnection_queue
- //!beause conenciton need to be reopend
- //reset real connection
- ci->conn = NULL;
- ci->ext_unit_conn->online = false;
- ci->ext_unit_conn->accepted_state = -1;
- //set retry timeout after five seconds
- ci->next_reconnection_retry_ts = TimingUtil::getTimestampWithDelay(5000, true);
+ //in this case concnretion info need to be put into reconnection_queue
+ //!beause conenciton need to be reopend
+ //reset real connection
+ CHAOS_ASSERT(conn_info->ext_unit_conn.get());
+ conn_info->ext_unit_conn->online = false;
+ conn_info->ext_unit_conn->accepted_state = -1;
+ //set retry timeout after five seconds
+ conn_info->next_reconnection_retry_ts = TimingUtil::getTimestampWithDelay(5000, true);
+ delete(conn_metadata);
+ break;
+ }
+
+ case MG_EV_POLL:{
+ //execute opcode for connection
+ ChaosWriteLock conn_wl(conn_info->smutex);
+ while(conn_info->opcode_queue.empty() == false) {
+ OpcodeShrdPtr op = conn_info->opcode_queue.front();
+ conn_info->opcode_queue.pop();
+ switch(op->op_type) {
+ case OpcodeInfoTypeSend: {
+ switch (op->data_opcode) {
+ case EUCMessageOpcodeWhole:
+ mg_send_websocket_frame(conn, WEBSOCKET_OP_TEXT, op->data->getBuffer(), op->data->getBufferSize());
+ break;
+ case EUCPhaseStartFragment:
+ mg_send_websocket_frame(conn, WEBSOCKET_OP_TEXT|WEBSOCKET_DONT_FIN, op->data->getBuffer(), op->data->getBufferSize());
+ break;
+ case EUCPhaseContinueFragment:
+ mg_send_websocket_frame(conn, WEBSOCKET_OP_TEXT|WEBSOCKET_DONT_FIN, op->data->getBuffer(), op->data->getBufferSize());
+ break;
+ case EUCPhaseEndFragment:
+ mg_send_websocket_frame(conn, WEBSOCKET_OP_TEXT, op->data->getBuffer(), op->data->getBufferSize());
+ break;
+ }
+ break;
+ }
+ default:{break;}
+ }
+ op->wait_termination_semaphore.unlock();
}
break;
}
}
}
-
-int HTTPClientAdapter::sendDataToConnection(const std::string& connection_identifier,
- chaos::common::data::CDBufferUniquePtr data,
- const EUCMessageOpcode opcode) {
- LOpcodeShrdPtrQueueWriteLock wconnl = post_evt_op_queue.getWriteLockObject();
- OpcodeShrdPtr op(new Opcode());
- op->identifier = connection_identifier;
- op->op_type = OpcodeInfoTypeSend;
- op->data = ChaosMoveOperator(data);
- op->data_opcode = opcode;
- post_evt_op_queue().push(op);
- return 0;
-}
-
-int HTTPClientAdapter::closeConnection(const std::string& connection_identifier) {
- LOpcodeShrdPtrQueueWriteLock wconnl = post_evt_op_queue.getWriteLockObject();
- OpcodeShrdPtr op(new Opcode());
- op->identifier = connection_identifier;
- op->op_type = OpcodeInfoTypeCloseConnection;
- post_evt_op_queue().push(op);
- return 0;
-}
diff --git a/chaos/common/external_unit/http_adapter/HTTPClientAdapter.h b/chaos/common/external_unit/http_adapter/HTTPClientAdapter.h
index 5748a0fa130b35c815a4946d86240208bb699d0c..57ba3d65ea94562153998f4da15fc45c8636c94e 100644
--- a/chaos/common/external_unit/http_adapter/HTTPClientAdapter.h
+++ b/chaos/common/external_unit/http_adapter/HTTPClientAdapter.h
@@ -37,35 +37,45 @@ namespace chaos {
namespace common {
namespace external_unit {
namespace http_adapter {
+
+ /**
+ * @brief websocket http client implementation for external unit
+ *
+ */
class HTTPClientAdapter:
public HTTPBaseAdapter,
public AbstractClientAdapter {
+ /**
+ * @brief Identify the conenction requested to
+ * the adapter
+ */
struct ConnectionInfo {
- std::string endpoint_url;
+ ChaosSharedMutex smutex;
+ const std::string endpoint_url;
uint64_t next_reconnection_retry_ts;
- ChaosSharedMutex smux;
- HTTPClientAdapter *class_instance;
- struct mg_connection *conn;
ChaosSharedPtr<ExternalUnitConnection> ext_unit_conn;
-
- ConnectionInfo():
+ //!opocode sent to real connection
+ OpcodeShrdPtrQueue opcode_queue;
+ ConnectionInfo(const std::string& _endpoint_url):
+ endpoint_url(_endpoint_url),
next_reconnection_retry_ts(0),
- class_instance(NULL),
- conn(NULL),
ext_unit_conn(){}
};
typedef ChaosSharedPtr<ConnectionInfo> ConnectionInfoShrdPtr;
+ CHAOS_DEFINE_LOCKABLE_OBJECT(OpcodeShrdPtrQueue, LOpcodeShrdPtrQueue);
- CHAOS_DEFINE_MAP_FOR_TYPE(std::string, ConnectionInfoShrdPtr, MapReconnectionInfo);
- CHAOS_DEFINE_LOCKABLE_OBJECT(MapReconnectionInfo, LMapReconnectionInfo);
+ //!associate connection identifier to the connection info
+ CHAOS_DEFINE_MAP_FOR_TYPE(std::string, ConnectionInfoShrdPtr, MapConnectionInfo);
+ CHAOS_DEFINE_LOCKABLE_OBJECT(MapConnectionInfo, LMapConnectionInfo);
bool run;
- struct mg_mgr mgr;
+
+ ChaosAtomic<uint32_t> message_broadcasted;
uint32_t poll_counter;
uint32_t rest_poll_time;
//!map that hold the connection to use
- LMapReconnectionInfo map_connection;
+ LMapConnectionInfo map_connection;
void poller();
void performReconnection();
diff --git a/chaos/common/external_unit/http_adapter/HTTPServerAdapter.cpp b/chaos/common/external_unit/http_adapter/HTTPServerAdapter.cpp
index b7f91b5073a924beb5529cb577481e0ae00965ac..dc56cfc7b3c20648c421ac4f56f290a4cef5b253 100644
--- a/chaos/common/external_unit/http_adapter/HTTPServerAdapter.cpp
+++ b/chaos/common/external_unit/http_adapter/HTTPServerAdapter.cpp
@@ -41,6 +41,7 @@ using namespace chaos::common::external_unit::http_adapter;
#define DBG DBG_LOG(HTTPServerAdapter)
#define ERR ERR_LOG(HTTPServerAdapter)
+#define LOCK_POST_EVENT_QUEUE LOpcodeShrdPtrQueueWriteLock wl_global_queue = post_evt_op_queue.getWriteLockObject()
HTTPServerAdapter::HTTPServerAdapter():
run(false),
@@ -67,14 +68,14 @@ void HTTPServerAdapter::init(void *init_data) throw (chaos::CException) {
const std::string http_port_str = boost::lexical_cast<std::string>(InetUtility::scanForLocalFreePort(boost::lexical_cast<int>(setting.publishing_port)));
root_connection = mg_bind(&mgr, http_port_str.c_str(), HTTPServerAdapter::eventHandler);
if(root_connection == NULL) {throw CException(-1, "Error creating http connection", __PRETTY_FUNCTION__);}
- root_connection->user_data = this;
-
+ root_connection->user_data = new ConnectionMetadata<HTTPServerAdapter>("", this);
+
mg_set_protocol_http_websocket(root_connection);
s_http_server_opts.document_root = ""; // Serve current directory
s_http_server_opts.enable_directory_listing = "no";
//
CObjectProcessingQueue<ServerWorkRequest>::init(setting.thread_number);
-
+
thread_poller.reset(new boost::thread(boost::bind(&HTTPServerAdapter::poller, this)));
}
@@ -91,145 +92,187 @@ void HTTPServerAdapter::poller() {
INFO << "Entering thread poller";
while (run) {
mg_mgr_poll(&mgr, 1);
-
+
//consume opcode queue
{
- LOpcodeShrdPtrQueueReadLock wconnl = post_evt_op_queue.getReadLockObject();
+ LOpcodeShrdPtrQueueReadLock post_queue_lock = post_evt_op_queue.getReadLockObject();
while(post_evt_op_queue().empty() == false) {
OpcodeShrdPtr op = post_evt_op_queue().front();
post_evt_op_queue().pop();
- wconnl->unlock();
- switch(op->op_type) {
- case OpcodeInfoTypeSend:{
- LMapConnectionReadLock wconnl = map_connection.getReadLockObject();
- if(!map_m_conn_ext_conn.hasRightKey(static_cast<std::string>(op->identifier))) break;
- mg_connection *nc = reinterpret_cast<mg_connection*>(map_m_conn_ext_conn.findByRightKey(static_cast<std::string>(op->identifier)));
- switch (op->data_opcode) {
- case EUCMessageOpcodeWhole:
- mg_send_websocket_frame(nc, WEBSOCKET_OP_TEXT, op->data->getBuffer(), op->data->getBufferSize());
- break;
- case EUCPhaseStartFragment:
- mg_send_websocket_frame(nc, WEBSOCKET_OP_TEXT|WEBSOCKET_DONT_FIN, op->data->getBuffer(), op->data->getBufferSize());
- break;
-
- case EUCPhaseContinueFragment:
- mg_send_websocket_frame(nc, WEBSOCKET_OP_TEXT|WEBSOCKET_DONT_FIN, op->data->getBuffer(), op->data->getBufferSize());
- break;
-
- case EUCPhaseEndFragment:
- mg_send_websocket_frame(nc, WEBSOCKET_OP_TEXT, op->data->getBuffer(), op->data->getBufferSize());
- break;
+ post_queue_lock->unlock();
+ {
+ LMapConnectionWriteLock map_connection_lock = map_connection.getWriteLockObject();
+ switch(op->op_type) {
+ case OpcodeInfoTypeCloseConnection: {
+ struct mg_connection *target_connection = findConnection<HTTPServerAdapter>(op->identifier);
+ if(target_connection) {
+ executeOpcodeOnConnection(op, target_connection);
+ }
+ //erase connection
+ map_connection().erase(op->identifier);
}
- break;
- }
- case OpcodeInfoTypeCloseConnection:{
- LMapConnectionWriteLock wconnl = map_connection.getWriteLockObject();
- if(!map_m_conn_ext_conn.hasRightKey(static_cast<std::string>(op->identifier))) break;
- mg_connection *nc = reinterpret_cast<mg_connection*>(map_m_conn_ext_conn.findByRightKey(static_cast<std::string>(op->identifier)));
- if(nc) {
- nc->flags |= MG_F_CLOSE_IMMEDIATELY;
+ case OpcodeInfoTypeSend: {
+ struct mg_connection *target_connection = findConnection<HTTPServerAdapter>(op->identifier);
+ if(target_connection) {
+ executeOpcodeOnConnection(op, target_connection);
+ }
+ break;
}
- map_connection().erase(reinterpret_cast<uintptr_t>(nc));
- map_m_conn_ext_conn.removebyLeftKey(reinterpret_cast<uintptr_t>(nc));
- break;
- }
-
- case OpcodeInfoTypeCloseConnectionForEndpoint:{
- LMapConnectionWriteLock wconnl = map_connection.getWriteLockObject();
- for(MapConnectionIterator it = map_connection().begin(),
- end = map_connection().end();
- it != end;) {
- if(it->second->getEndpointIdentifier().compare(op->identifier) == 0) {
- if(map_m_conn_ext_conn.hasRightKey(static_cast<std::string>(it->second->connection_identifier))) {
- mg_connection *nc = reinterpret_cast<mg_connection*>(map_m_conn_ext_conn.findByRightKey(static_cast<std::string>(it->second->connection_identifier)));
- if(nc) {
- nc->flags |= MG_F_CLOSE_IMMEDIATELY;
- }
- //remove connection
+ case OpcodeInfoTypeCloseConnectionForEndpoint:{
+ for(MapConnectionIterator it = map_connection().begin(),
+ end = map_connection().end();
+ it != end;) {
+ //!op->identifier now is the identifier of the enpoint
+ if(it->second->getEndpointIdentifier().compare(op->identifier) == 0) {
+ //connection below to the endpoint that need to be removed
+ const std::string connection_uuid = it->second->connection_identifier;
+ struct mg_connection *taget_connection = findConnection<HTTPServerAdapter>(connection_uuid);
+ op->identifier = connection_uuid;
+ op->op_type = OpcodeInfoTypeCloseConnection;
+ executeOpcodeOnConnection(op, taget_connection);
map_connection().erase(it++);
+ } else {
+ ++it;
}
- } else {
- ++it;
}
+ break;
}
- break;
}
}
op->wait_termination_semaphore.unlock();
- wconnl->lock();
+ post_queue_lock->lock();
}
}
}
INFO << "Leaving thread poller";
}
+void HTTPServerAdapter::executeOpcodeOnConnection(OpcodeShrdPtr op, mg_connection *target_connection) {
+ switch(op->op_type) {
+ case OpcodeInfoTypeCloseConnection:{
+ //!op->identifier now is the identifier of the external connection
+ //search real connection
+ ConnectionMetadata<HTTPServerAdapter> *conn_metadata = static_cast<ConnectionMetadata<HTTPServerAdapter> * >(target_connection->user_data);
+ target_connection->flags |= MG_F_SEND_AND_CLOSE;
+ delete(conn_metadata);
+ target_connection->user_data = NULL;
+ break;
+ }
+
+ case OpcodeInfoTypeSend:{
+ switch (op->data_opcode) {
+ case EUCMessageOpcodeWhole:
+ mg_send_websocket_frame(target_connection, WEBSOCKET_OP_TEXT, op->data->getBuffer(), op->data->getBufferSize());
+ break;
+ case EUCPhaseStartFragment:
+ mg_send_websocket_frame(target_connection, WEBSOCKET_OP_TEXT|WEBSOCKET_DONT_FIN, op->data->getBuffer(), op->data->getBufferSize());
+ break;
+
+ case EUCPhaseContinueFragment:
+ mg_send_websocket_frame(target_connection, WEBSOCKET_OP_TEXT|WEBSOCKET_DONT_FIN, op->data->getBuffer(), op->data->getBufferSize());
+ break;
+
+ case EUCPhaseEndFragment:
+ mg_send_websocket_frame(target_connection, WEBSOCKET_OP_TEXT, op->data->getBuffer(), op->data->getBufferSize());
+ break;
+ }
+ break;
+ }
+ default:{break;}
+ }
+}
+
const std::string HTTPServerAdapter::getSerializationType(http_message *http_message) {
- CHAOS_ASSERT(http_message);
struct mg_str *value = mg_get_http_header(http_message, "Content-Type");
if(value == NULL) {
value = mg_get_http_header(http_message, "content-type");
if(value == NULL) return "";
}
-
+
std::string ser_type(value->p, value->len);
std::transform(ser_type.begin(), ser_type.end(), ser_type.begin(), ::tolower);
return ser_type;
}
-void HTTPServerAdapter::manageWSHandshake(ServerWorkRequest& wr) {
+void HTTPServerAdapter::manageWSHandshake(mg_connection *nc,
+ http_message *message) {
+
+ const std::string s_type = getSerializationType(message);
+ if(s_type.size() == 0) {
+ std::string error = "Serialization type not found";
+ mg_send_head(nc, 400, error.size(), "Content-Type: text/plain");
+ mg_printf(nc, "%s", error.c_str());
+ return;
+ }
+
char addr[32];
- mg_sock_addr_to_str(&wr.nc->sa, addr, sizeof(addr),
+ std::string uri(message->uri.p, message->uri.len);
+ mg_sock_addr_to_str(&nc->sa, addr, sizeof(addr),
MG_SOCK_STRINGIFY_IP | MG_SOCK_STRINGIFY_PORT);
- INFO << CHAOS_FORMAT("Received new connection for endoint %1% from %2%", %wr.uri%addr);
- LMapEndpointReadLock wl = map_endpoint.getReadLockObject();
- MapEndpointIterator endpoint_it = map_endpoint().find(wr.uri);
+ INFO << CHAOS_FORMAT("Received new connection for endoint %1% from %2%", %uri%addr);
+ //lock global queue
+ MapEndpointIterator endpoint_it = map_endpoint().find(uri);
if(endpoint_it == map_endpoint().end()) {
- sendWSJSONError(wr.nc,
- -1,
- CHAOS_FORMAT("No endpoint found for '%1%'", %wr.uri),
- false);
- sendWSJSONAcceptedConnection(wr.nc,
- false,
- true);
+ LOpcodeShrdPtrQueueWriteLock wl_global_queue = post_evt_op_queue.getWriteLockObject();
+ const std::string conn_uuid = UUIDUtil::generateUUIDLite();
+ nc->user_data = new ConnectionMetadata<HTTPServerAdapter>(conn_uuid, this);
+ post_evt_op_queue().push(composeJSONErrorResposneOpcode(conn_uuid,
+ -1,
+ CHAOS_FORMAT("No endpoint found for '%1%'", %uri)));
+
+ post_evt_op_queue().push(composeAcceptOpcode(conn_uuid, false));
+ post_evt_op_queue().push(composeCloseOpcode(conn_uuid));
+
return;
}
-
+
//check if endpoint can accept more connection
if(endpoint_it->second->canAcceptMoreConnection() == false) {
- //write error for no more connection accepted by endpoint
- sendWSJSONError(wr.nc,
- -2,
- CHAOS_FORMAT("No more connection accepted by endpoint '%1%'", %wr.uri),
- false);
- sendWSJSONAcceptedConnection(wr.nc,
- false,
- true);
+ LOCK_POST_EVENT_QUEUE;
+ const std::string conn_uuid = UUIDUtil::generateUUIDLite();
+ nc->user_data = new ConnectionMetadata<HTTPServerAdapter>(conn_uuid, this);
+ post_evt_op_queue().push(composeJSONErrorResposneOpcode(conn_uuid,
+ -1,
+ CHAOS_FORMAT("No more connection accepted by endpoint '%1%'", %uri)));
+
+ post_evt_op_queue().push(composeAcceptOpcode(conn_uuid, false));
+ post_evt_op_queue().push(composeCloseOpcode(conn_uuid));
} else {
//get instance for serializer
- ChaosUniquePtr<serialization::AbstractExternalSerialization> serializer = ExternalUnitManager::getInstance()->getNewSerializationInstanceForType(wr.s_type);
+ ChaosUniquePtr<serialization::AbstractExternalSerialization> serializer = ExternalUnitManager::getInstance()->getNewSerializationInstanceForType(s_type);
if(!serializer.get()) {
- //write error for no more connection accepted by endpoint
- sendWSJSONError(wr.nc,
- -3,
- CHAOS_FORMAT("Unable to find the serialization plugin for '%1%'", %wr.s_type),
- false);
- sendWSJSONAcceptedConnection(wr.nc,
- false,
- true);
+ LOCK_POST_EVENT_QUEUE;
+ const std::string conn_uuid = UUIDUtil::generateUUIDLite();
+ nc->user_data = new ConnectionMetadata<HTTPServerAdapter>(conn_uuid, this);
+ post_evt_op_queue().push(composeJSONErrorResposneOpcode(conn_uuid,
+ -1,
+ CHAOS_FORMAT("Unable to find the serialization plugin for '%1%'", %uri)));
+
+ post_evt_op_queue().push(composeAcceptOpcode(conn_uuid, false));
+ post_evt_op_queue().push(composeCloseOpcode(conn_uuid));
} else {
//we can create a new connection
ChaosSharedPtr<ExternalUnitConnection> conn_ptr(new ExternalUnitConnection(this,
endpoint_it->second,
ChaosMoveOperator(serializer)));
+ ChaosUniquePtr< ConnectionMetadata<HTTPServerAdapter> > connection_metadata = ChaosUniquePtr< ConnectionMetadata<HTTPServerAdapter> >(new ConnectionMetadata<HTTPServerAdapter>(conn_ptr->connection_identifier, this));
LMapConnectionWriteLock wconnl = map_connection.getWriteLockObject();
- map_connection().insert(MapConnectionPair(reinterpret_cast<uintptr_t>(wr.nc),
- conn_ptr));
- //add the mapping from mongoose conenciton and ext unit one
- map_m_conn_ext_conn.insert(reinterpret_cast<uintptr_t>(wr.nc),
- conn_ptr->connection_identifier);
- sendWSJSONAcceptedConnection(wr.nc,
- true,
- false);
+ std::pair<MapConnectionIterator,bool> conn_pair = map_connection().insert(MapConnectionPair(conn_ptr->connection_identifier,
+ conn_ptr));
+ nc->user_data = connection_metadata.release();
+ if(conn_pair.second) {
+ LOCK_POST_EVENT_QUEUE;
+ //all is gone well so we need to associate the connection metadata to the real connection
+ post_evt_op_queue().push(composeAcceptOpcode(conn_ptr->connection_identifier, true));
+ } else {
+ LOCK_POST_EVENT_QUEUE;
+ post_evt_op_queue().push(composeJSONErrorResposneOpcode(conn_ptr->connection_identifier,
+ -1,
+ CHAOS_FORMAT("Errore registering the new connection for '%1%'", %uri)));
+
+ post_evt_op_queue().push(composeAcceptOpcode(conn_ptr->connection_identifier, false));
+ post_evt_op_queue().push(composeCloseOpcode(conn_ptr->connection_identifier));
+ }
}
}
}
@@ -237,108 +280,85 @@ void HTTPServerAdapter::manageWSHandshake(ServerWorkRequest& wr) {
void HTTPServerAdapter::processBufferElement(ServerWorkRequest *request,
ElementManagingPolicy& policy) throw(CException) {
switch(request->r_type) {
- case WorkRequestTypeHttpRequest: {
- //http_message *message = static_cast<http_message*>(wr.message);
- mg_printf(request->nc, "%s", "HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n");
- mg_printf_http_chunk(request->nc, "!CHAOS Control External gateway not support http get or post");
- mg_send_http_chunk(request->nc, "", 0); /* Send empty chunk, the end of response */
- break;
- }
- case WorkRequestTypeWSHandshakeRequest: {
- manageWSHandshake(*request);
- break;
- }
case WorkRequestTypeWSFrame: {
int err = 0;
LMapConnectionReadLock wconnl = map_connection.getReadLockObject();
- if(map_connection().count(reinterpret_cast<uintptr_t>(request->nc))) {
- if((err = sendDataToEndpoint(*map_connection()[reinterpret_cast<uintptr_t>(request->nc)],
- ChaosMoveOperator(request->buffer)))) {
- //weh don't have found the sriealizer
- const std::string error = CHAOS_FORMAT("{error:%1%,message:\"%2%\"}", %err%map_connection()[reinterpret_cast<uintptr_t>(request->nc)]->getEndpointIdentifier());
- mg_send_websocket_frame(request->nc, WEBSOCKET_OP_TEXT, error.c_str(), error.size());
- }
+ MapConnectionIterator conn_it = map_connection().find(request->connection_uuid);
+ if(conn_it == map_connection().end()) {
+ return;
+ }
+ if((err = sendDataToEndpoint(*conn_it->second,
+ ChaosMoveOperator(request->buffer)))) {
+ //add error message to the queue
+ // const std::string error = CHAOS_FORMAT("{error:%1%,message:\"%2%\"}", %err%map_connection()[reinterpret_cast<uintptr_t>(request->nc)]->getEndpointIdentifier());
+ // mg_send_websocket_frame(request->nc, WEBSOCKET_OP_TEXT, error.c_str(), error.size());
}
break;
}
- case WorkRequestTypeWSCloseEvent: {
- LMapConnectionWriteLock wconnl = map_connection.getWriteLockObject();
- map_connection().erase(reinterpret_cast<uintptr_t>(request->nc));
- map_m_conn_ext_conn.removebyLeftKey(reinterpret_cast<uintptr_t>(request->nc));
- break;
- }
-
default:{break;}
}
}
void HTTPServerAdapter::eventHandler(mg_connection *nc, int ev, void *ev_data) {
- (void) nc;
- (void) ev_data;
- HTTPServerAdapter *adapter = static_cast<HTTPServerAdapter*>(nc->user_data);
- if(!adapter->run) return;
+ ConnectionMetadata<HTTPServerAdapter> *connection_metadata = static_cast< ConnectionMetadata<HTTPServerAdapter>* >(nc->user_data);
switch (ev) {
case MG_EV_ACCEPT:{
+ //new conenction has been accepted
break;
}
case MG_EV_HTTP_REQUEST: {
- http_message *message = static_cast<http_message*>(ev_data);
- ServerWorkRequest *req = new ServerWorkRequest(message->message.p,
- (uint32_t)message->message.len);
- req->r_type = WorkRequestTypeHttpRequest;
- req->s_type = getSerializationType(message);
- req->nc = nc;
- req->uri.assign(message->uri.p, message->uri.len);
- adapter->push(req);
+ mg_printf(nc, "%s", "HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n");
+ mg_printf_http_chunk(nc, "!CHAOS Control External gateway not support http get or post");
+ mg_send_http_chunk(nc, "", 0); /* Send empty chunk, the end of response */
+
break;
}
case MG_EV_WEBSOCKET_HANDSHAKE_REQUEST: {
- http_message *message = static_cast<http_message*>(ev_data);
- ChaosUniquePtr<ServerWorkRequest> req(new ServerWorkRequest(message->message.p,
- (uint32_t)message->message.len));
- req->r_type = WorkRequestTypeWSHandshakeRequest;
- req->s_type = getSerializationType(message);
- req->nc = nc;
- req->uri.assign(message->uri.p, message->uri.len);
- if(req->s_type.size() == 0) {
- std::string error = "Serialization type not found";
- mg_send_head(nc, 400, error.size(), "Content-Type: text/plain");
- mg_printf(nc, "%s", error.c_str());
- }else {
- adapter->push(req.release());
- }
+ connection_metadata->class_instance->manageWSHandshake(nc, static_cast<http_message*>(ev_data));
+
break;
}
case MG_EV_WEBSOCKET_FRAME: {
websocket_message *message = static_cast<websocket_message*>(ev_data);
- ServerWorkRequest *req = new ServerWorkRequest((const char *)message->data,
+ ServerWorkRequest *req = new ServerWorkRequest(connection_metadata->conn_uuid,
+ (const char *)message->data,
(uint32_t)message->size);
- req->r_type = WorkRequestTypeWSFrame;
- req->nc = nc;
- adapter->push(req);
+ connection_metadata->class_instance->push(req);
break;
}
+
case MG_EV_CLOSE:{
-// ServerWorkRequest *req = new ServerWorkRequest();
-// req->r_type = WorkRequestTypeWSCloseEvent;
-// req->nc = nc;
-// adapter->push(req);
- LMapConnectionWriteLock wconnl = adapter->map_connection.getWriteLockObject();
- adapter->map_connection().erase(reinterpret_cast<uintptr_t>(nc));
- adapter->map_m_conn_ext_conn.removebyLeftKey(reinterpret_cast<uintptr_t>(nc));
+ if(connection_metadata) {
+ LMapConnectionWriteLock wconnl = connection_metadata->class_instance->map_connection.getWriteLockObject();
+ connection_metadata->class_instance->map_connection().erase(connection_metadata->conn_uuid);
+ delete(connection_metadata);
+ nc->user_data = NULL;
+ }
break;
}
+
+ case MG_EV_POLL: break;
}
}
-void HTTPServerAdapter::sendWSJSONAcceptedConnection(mg_connection *nc,
- bool accepted,
- bool close_connection) {
+OpcodeShrdPtr HTTPServerAdapter::composeAcceptOpcode(const std::string& connection_uuid,
+ bool accepted) {
chaos::common::data::CDataWrapper err_data_pack;
err_data_pack.addInt32Value("accepted_connection", accepted);
const std::string accepted_json = err_data_pack.getCompliantJSONString();
- mg_send_websocket_frame(nc, WEBSOCKET_OP_TEXT, accepted_json.c_str(), accepted_json.size());
- if(close_connection){mg_send_websocket_frame(nc, WEBSOCKET_OP_CLOSE, NULL, 0);}
+ OpcodeShrdPtr op(new Opcode());
+ op->identifier = connection_uuid;
+ op->op_type = OpcodeInfoTypeSend;
+ op->data = CDBufferUniquePtr(new CDataBuffer(accepted_json.c_str(), (uint32_t)accepted_json.size()));
+ op->data_opcode = EUCMessageOpcodeWhole;
+ return op;
+}
+
+OpcodeShrdPtr HTTPServerAdapter::composeCloseOpcode(const std::string& connection_uuid) {
+ OpcodeShrdPtr op(new Opcode());
+ op->identifier = connection_uuid;
+ op->op_type = OpcodeInfoTypeCloseConnection;
+ return op;
}
int HTTPServerAdapter::registerEndpoint(ExternalUnitServerEndpoint& endpoint) {
@@ -351,31 +371,41 @@ int HTTPServerAdapter::registerEndpoint(ExternalUnitServerEndpoint& endpoint) {
int HTTPServerAdapter::deregisterEndpoint(ExternalUnitServerEndpoint& endpoint) {
//lock for write conenction and endpoint
LMapEndpointWriteLock wl = map_endpoint.getWriteLockObject();
+ LMapConnectionReadLock mcwl = map_connection.getReadLockObject();
MapEndpointIterator me_it = map_endpoint().find(endpoint.getIdentifier());
if(me_it == map_endpoint().end()) return 0;
- me_it->second = NULL;
- map_endpoint().erase(me_it);
- OpcodeShrdPtr op(new Opcode());
- // map_endpoint().erase(endpoint.getIdentifier());
//at this point no new conneciton can be associated to the endpoint
- {
- LOpcodeShrdPtrQueueWriteLock wconnl = post_evt_op_queue.getWriteLockObject();
- op->identifier = endpoint.getIdentifier();
- op->op_type = OpcodeInfoTypeCloseConnectionForEndpoint;
- post_evt_op_queue().push(op);
- }
- //we need to wait that opcode has terminated
- op->wait_termination_semaphore.wait();
+ me_it->second = NULL;
+ //scan all conenciton and push opcode for close it
+ for(MapConnectionIterator it = map_connection().begin(),
+ end = map_connection().end();
+ it != end;
+ it++) {
+ //!op->identifier now is the identifier of the enpoint
+ if(it->second->getEndpointIdentifier().compare(endpoint.getIdentifier()) == 0) {
+ //connection below to the endpoint that need to be removed
+ closeConnection(it->second->connection_identifier);
+ }
+ }
+ map_endpoint().erase(me_it);
return 0;
}
int HTTPServerAdapter::sendDataToConnection(const std::string& connection_identifier,
chaos::common::data::CDBufferUniquePtr data,
const EUCMessageOpcode opcode) {
+ return _sendDataToConnectionQueue(connection_identifier,
+ ChaosMoveOperator(data),
+ opcode);
+}
+
+int HTTPServerAdapter::_sendDataToConnectionQueue(const std::string& conn_uuid,
+ chaos::common::data::CDBufferUniquePtr data,
+ const EUCMessageOpcode opcode) {
LOpcodeShrdPtrQueueWriteLock wconnl = post_evt_op_queue.getWriteLockObject();
OpcodeShrdPtr op(new Opcode());
- op->identifier = connection_identifier;
+ op->identifier = conn_uuid;
op->op_type = OpcodeInfoTypeSend;
op->data = ChaosMoveOperator(data);
op->data_opcode = opcode;
@@ -385,9 +415,6 @@ int HTTPServerAdapter::sendDataToConnection(const std::string& connection_identi
int HTTPServerAdapter::closeConnection(const std::string& connection_identifier) {
LOpcodeShrdPtrQueueWriteLock wconnl = post_evt_op_queue.getWriteLockObject();
- OpcodeShrdPtr op(new Opcode());
- op->identifier = connection_identifier;
- op->op_type = OpcodeInfoTypeCloseConnection;
- post_evt_op_queue().push(op);
+ post_evt_op_queue().push(composeCloseOpcode(connection_identifier));
return 0;
}
diff --git a/chaos/common/external_unit/http_adapter/HTTPServerAdapter.h b/chaos/common/external_unit/http_adapter/HTTPServerAdapter.h
index 0fa0a4ba66963e9751169b4b5923bc3116be51d5..1fa647eb6ba1629d4a3642e068179696404a8abe 100644
--- a/chaos/common/external_unit/http_adapter/HTTPServerAdapter.h
+++ b/chaos/common/external_unit/http_adapter/HTTPServerAdapter.h
@@ -29,24 +29,41 @@ namespace chaos{
namespace common {
namespace external_unit {
namespace http_adapter {
-
+
+ struct ServerWorkRequest {
+ const WorkRequestType r_type;
+ const std::string connection_uuid;
+ ChaosUniquePtr<chaos::common::data::CDataBuffer> buffer;
+
+ ServerWorkRequest():
+ r_type(WorkRequestTypeUnspecified),
+ buffer(){}
+
+ ServerWorkRequest(const std::string& _connection_uuid,
+ const char *ptr,
+ uint32_t size):
+ r_type(WorkRequestTypeWSFrame),
+ connection_uuid(_connection_uuid),
+ buffer(new chaos::common::data::CDataBuffer(ptr, size)){}
+ };
+
+
+ typedef ChaosSharedPtr<ServerWorkRequest> ServerWorkRequestShrdPtr;
//!External gateway root class
class HTTPServerAdapter:
public AbstractServerAdapter,
protected CObjectProcessingQueue< ServerWorkRequest >,
public HTTPBaseAdapter {
friend class ExternalUnitConnection;
- CHAOS_DEFINE_MAP_FOR_TYPE(uintptr_t, ChaosSharedPtr<ExternalUnitConnection>, MapConnection);
+ CHAOS_DEFINE_MAP_FOR_TYPE(std::string, ChaosSharedPtr<ExternalUnitConnection>, MapConnection);
CHAOS_DEFINE_LOCKABLE_OBJECT(MapConnection, LMapConnection);
//!contains all connection
LMapConnection map_connection;
- chaos::common::utility::Bimap<uintptr_t, std::string> map_m_conn_ext_conn;
bool run;
HTTPServerAdapterSetting setting;
- struct mg_mgr mgr;
struct mg_connection *root_connection;
struct mg_serve_http_opts s_http_server_opts;
@@ -55,10 +72,17 @@ namespace chaos{
static void eventHandler(mg_connection *nc,
int ev,
void *ev_data);
- void manageWSHandshake(ServerWorkRequest& wr);
- void sendWSJSONAcceptedConnection(mg_connection *nc,
- bool accepted,
- bool close_connection);
+ void manageWSHandshake(mg_connection *nc,
+ http_message *message);
+ OpcodeShrdPtr composeAcceptOpcode(const std::string& connection_uuid,
+ bool accepted);
+ OpcodeShrdPtr composeCloseOpcode(const std::string& connection_uuid);
+
+ void consumeConenctionMessageQueue(mg_connection *nc);
+ int _sendDataToConnectionQueue(const std::string& conn_uuid,
+ chaos::common::data::CDBufferUniquePtr data,
+ const EUCMessageOpcode opcode);
+ void executeOpcodeOnConnection(OpcodeShrdPtr op, mg_connection *nc);
protected:
void processBufferElement(ServerWorkRequest *request, ElementManagingPolicy& policy) throw(CException);
int sendDataToConnection(const std::string& connection_identifier,
diff --git a/chaos/common/external_unit/http_adapter/http_adapter_types.h b/chaos/common/external_unit/http_adapter/http_adapter_types.h
index 91cc44c504f3e206a1c92eac064ade2678d66e97..6d718c5a30f93653a73efea9a05d185c2534a940 100644
--- a/chaos/common/external_unit/http_adapter/http_adapter_types.h
+++ b/chaos/common/external_unit/http_adapter/http_adapter_types.h
@@ -32,10 +32,7 @@ namespace chaos{
namespace http_adapter {
typedef enum {
WorkRequestTypeUnspecified,
- WorkRequestTypeHttpRequest,
- WorkRequestTypeWSHandshakeRequest,
- WorkRequestTypeWSFrame,
- WorkRequestTypeWSCloseEvent
+ WorkRequestTypeWSFrame
} WorkRequestType;
struct HTTPServerAdapterSetting {
diff --git a/chaos/common/network/PerformanceManagment.cpp b/chaos/common/network/PerformanceManagment.cpp
index ac775d561ee1647ff4bb4afb6f05ec8cc460fe07..1834412ba8d8308aec4b18a2462342127ee17c0f 100644
--- a/chaos/common/network/PerformanceManagment.cpp
+++ b/chaos/common/network/PerformanceManagment.cpp
@@ -86,7 +86,7 @@ void PerformanceManagment::stop() throw(chaos::CException) {
}
DirectIOClient *PerformanceManagment::getLocalDirectIOClientInstance() {
- boost::unique_lock<boost::mutex>(mutext_client_connection);
+ boost::unique_lock<boost::mutex> ul(mutext_client_connection);
if(!global_performance_connection) {
global_performance_connection = network_broker->getSharedDirectIOClientInstance();
//if(!global_performance_connection) throw chaos::CException(-1, "Performance direct io client creation error", __PRETTY_FUNCTION__);
diff --git a/chaos/common/pool/ResourcePool.h b/chaos/common/pool/ResourcePool.h
index dc73f494201d17eecff874e8d4a562976e6f7d23..82924d49869dcbe7427c86dce0539f4fcc7e5322 100644
--- a/chaos/common/pool/ResourcePool.h
+++ b/chaos/common/pool/ResourcePool.h
@@ -196,7 +196,6 @@ delete(x);
Performa a new resource allocation and push new one on pool queue
*/
inline void _pushNewResourceinPool() {
- bool success = false;
uint32_t alive_for_ms = 0;
//create temporare autoPtr for safe operation in case of exception
diff --git a/chaos/common/utility/InetUtility.cpp b/chaos/common/utility/InetUtility.cpp
index 438ddbdef8887c9509323b7928e74d6a2230d9ae..76df6db4cba84063c2538e8d69e1142123d24c46 100644
--- a/chaos/common/utility/InetUtility.cpp
+++ b/chaos/common/utility/InetUtility.cpp
@@ -124,7 +124,6 @@ void InetUtility::checkInterfaceName(std::vector<InterfaceInfo>& interface_infos
}
bool InetUtility::checkWellFormedHostPort(std::string host_port) {
- bool ret=false;
boost::system::error_code ec;
boost::asio::ip::address::from_string( host_port, ec );
if ( ec )
@@ -133,7 +132,6 @@ bool InetUtility::checkWellFormedHostPort(std::string host_port) {
}
bool InetUtility::checkWellFormedHostNamePort(std::string host_port) {
- bool ret=false;
int pos=host_port.find(":");
if(pos == std::string::npos){
return false;
diff --git a/config/CMakeChaos.txt b/config/CMakeChaos.txt
index 94e39dffc933beb5f94fd90ad75a5244ea695cfc..2aa3bfed9091adcc5f9b99b81f5cec258906e975 100644
--- a/config/CMakeChaos.txt
+++ b/config/CMakeChaos.txt
@@ -64,13 +64,24 @@ ADD_DEFINITIONS(-DCSLIB_BUILD_ID=${CHAOS_BUILD_ID})
if(CMAKE_BUILD_TYPE MATCHES PROFILE)
MESG("ENABLING PROFILE")
- set (CHAOS_CXX_FLAGS "${CHAOS_CXX_FLAGS} -g -fprofile-arcs -ftest-coverage")
- set (CHAOS_C_FLAGS "${CHAOS_C_FLAGS} -g -fprofile-arcs -ftest-coverage")
+ set(CHAOS_DEBUG ON)
+ set (CHAOS_CXX_FLAGS "${CHAOS_CXX_FLAGS} -fprofile-arcs -ftest-coverage")
+ set (CHAOS_C_FLAGS "${CHAOS_C_FLAGS} -fprofile-arcs -ftest-coverage")
set (CHAOS_LINKER_FLAGS "${CHAOS_LINKER_FLAGS} -fprofile-arcs -ftest-coverage")
ELSE()
MESG("BUILD TYPE:${CMAKE_BUILD_TYPE}")
ENDIF()
+if(CMAKE_BUILD_TYPE MATCHES DEBUG)
+ set(CHAOS_DEBUG ON)
+ENDIF()
+
+if (CHAOS_DEBUG)
+ MESG("Enable debug mode")
+ set (CHAOS_CXX_FLAGS "${CHAOS_CXX_FLAGS} -DDEBUG -g")
+ set (CHAOS_C_FLAGS "${CHAOS_C_FLAGS} -DDEBUG -g")
+ SET(CMAKE_BUILD_TYPE Debug)
+ENDIF()
set (BOOST_CXX_ADDITIONAL_FLAGS "")
IF(CHAOS_ENABLE_C11 AND NOT CHAOS_TARGET)
@@ -120,7 +131,7 @@ IF(${SYSTEM_NAME} MATCHES "Darwin")
SET(CHAOS_BOOST_FLAGS toolset=clang)
CheckCompiler("CXX" clang++)
CheckCompiler("C" clang)
- SET(CHAOS_CXX_FLAGS "${CHAOS_CXX_FLAGS} -std=c++11 -stdlib=libc++")
+ SET(CHAOS_CXX_FLAGS "${CHAOS_CXX_FLAGS} -stdlib=libc++")
SET(CHAOS_C_FLAGS "${CHAOS_C_FLAGS} -stdlib=libc++")
set(CHAOS_LINKER_FLAGS "-stdlib=libc++")
@@ -289,20 +300,15 @@ INCLUDE_DIRECTORIES(${CHAOS_FRAMEWORK})
# INCLUDE_DIRECTORIES($ENV{CHAOS_BUNDLE}/chaosframework)
# ENDIF()
-If(CHAOS_C_COMPILER)
- MESG("Setting C compiler ${CHAOS_C_COMPILER}")
- SET(CMAKE_C_COMPILER ${CHAOS_C_COMPILER})
-ENDIF()
+#If(CHAOS_C_COMPILER)
+# MESG("Setting C compiler ${CHAOS_C_COMPILER}")
+# SET(CMAKE_C_COMPILER ${CHAOS_C_COMPILER})
+#ENDIF()
SET( CMAKE_CHAOS $ENV{CHAOS_CMAKE_FLAGS})
#ADD_DEFINITIONS(-O2)
-if (CHAOS_DEBUG)
- add_definitions(-DDEBUG -g)
- SET(CMAKE_BUILD_TYPE Debug)
-ENDIF()
-
#set(FrameworkLib $ENV{CHAOS_LINK_LIBRARY})
#separate_arguments(FrameworkLib)