http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7528d23e/thirdparty/librdkafka-0.11.1/packaging/nuget/templates/librdkafka.redist.nuspec ---------------------------------------------------------------------- diff --git a/thirdparty/librdkafka-0.11.1/packaging/nuget/templates/librdkafka.redist.nuspec b/thirdparty/librdkafka-0.11.1/packaging/nuget/templates/librdkafka.redist.nuspec deleted file mode 100644 index f48e523..0000000 --- a/thirdparty/librdkafka-0.11.1/packaging/nuget/templates/librdkafka.redist.nuspec +++ /dev/null @@ -1,21 +0,0 @@ -<?xml version="1.0" encoding="utf-8"?> -<package xmlns="http://schemas.microsoft.com/packaging/2011/10/nuspec.xsd";> - <metadata> - <id>librdkafka.redist</id> - <version>${version}</version> - <title>librdkafka - redistributable</title> - <authors>Magnus Edenhill, edenhill</authors> - <owners>Magnus Edenhill, edenhill</owners> - <requireLicenseAcceptance>false</requireLicenseAcceptance> - <licenseUrl>https://github.com/edenhill/librdkafka/blob/master/LICENSES.txt</licenseUrl> - <projectUrl>https://github.com/edenhill/librdkafka</projectUrl> - <description>The Apache Kafka C/C++ client library - redistributable</description> - <summary>The Apache Kafka C/C++ client library</summary> - <releaseNotes>Release of librdkafka</releaseNotes> - <copyright>Copyright 2012-2017</copyright> - <tags>native apache kafka librdkafka C C++ nativepackage</tags> - </metadata> - <files> - <file src="**" /> - </files> -</package>
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7528d23e/thirdparty/librdkafka-0.11.1/packaging/nuget/templates/librdkafka.redist.props ---------------------------------------------------------------------- diff --git a/thirdparty/librdkafka-0.11.1/packaging/nuget/templates/librdkafka.redist.props b/thirdparty/librdkafka-0.11.1/packaging/nuget/templates/librdkafka.redist.props deleted file mode 100644 index 0b96886..0000000 --- a/thirdparty/librdkafka-0.11.1/packaging/nuget/templates/librdkafka.redist.props +++ /dev/null @@ -1,18 +0,0 @@ -<?xml version="1.0" encoding="utf-8" ?> -<Project ToolsVersion="12.0" xmlns="http://schemas.microsoft.com/developer/msbuild/2003";> - <ItemGroup> - <Content Include="$(MSBuildThisFileDirectory)..\..\runtimes\win7-x86\native\*"> - <Link>x86\%(Filename)%(Extension)</Link> - <CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory> - </Content> - <Content Include="$(MSBuildThisFileDirectory)..\..\runtimes\win7-x64\native\*"> - <Link>x64\%(Filename)%(Extension)</Link> - <CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory> - </Content> - </ItemGroup> - <ItemDefinitionGroup> - <ClCompile> - <AdditionalIncludeDirectories>$(MSBuildThisFileDirectory)include;%(AdditionalIncludeDirectories)</AdditionalIncludeDirectories> - </ClCompile> - </ItemDefinitionGroup> -</Project> http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7528d23e/thirdparty/librdkafka-0.11.1/packaging/nuget/templates/librdkafka.redist.targets ---------------------------------------------------------------------- diff --git a/thirdparty/librdkafka-0.11.1/packaging/nuget/templates/librdkafka.redist.targets b/thirdparty/librdkafka-0.11.1/packaging/nuget/templates/librdkafka.redist.targets deleted file mode 100644 index 4b5c9c4..0000000 --- a/thirdparty/librdkafka-0.11.1/packaging/nuget/templates/librdkafka.redist.targets +++ /dev/null @@ -1,19 +0,0 @@ -<Project ToolsVersion="4.0" xmlns="http://schemas.microsoft.com/developer/msbuild/2003";> - <ItemDefinitionGroup> - <Link> - <AdditionalDependencies Condition="'$(Platform)' == 'x64'">$(MSBuildThisFileDirectory)..\..\runtimes\win7-x64\native\librdkafka.lib;%(AdditionalDependencies)</AdditionalDependencies> - <AdditionalDependencies Condition="'$(Platform)' != 'x64'">$(MSBuildThisFileDirectory)..\..\runtimes\win7-x86\native\librdkafka.lib;%(AdditionalDependencies)</AdditionalDependencies> - <AdditionalLibraryDirectories Condition="'$(Platform)' == 'x64'">$(MSBuildThisFileDirectory)..\..\runtimes\win7-x64\native;%(AdditionalLibraryDirectories)</AdditionalLibraryDirectories> - <AdditionalLibraryDirectories Condition="'$(Platform)' != 'x64'">$(MSBuildThisFileDirectory)..\..\runtimes\win7-x86\native;%(AdditionalLibraryDirectories)</AdditionalLibraryDirectories> - </Link> - <ClCompile> - <AdditionalIncludeDirectories>$(MSBuildThisFileDirectory)include;%(AdditionalIncludeDirectories)</AdditionalIncludeDirectories> - </ClCompile> - </ItemDefinitionGroup> - <ItemGroup Condition="'$(Platform)' == 'x64'"> - <ReferenceCopyLocalPaths Include="$(MSBuildThisFileDirectory)..\..\runtimes\win7-x64\native\librdkafka.dll" /> - </ItemGroup> - <ItemGroup Condition="'$(Platform)' != 'x64'"> - <ReferenceCopyLocalPaths Include="$(MSBuildThisFileDirectory)..\..\runtimes\win7-x86\native\librdkafka.dll" /> - </ItemGroup> -</Project> http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7528d23e/thirdparty/librdkafka-0.11.1/packaging/nuget/zfile/__init__.py ---------------------------------------------------------------------- diff --git a/thirdparty/librdkafka-0.11.1/packaging/nuget/zfile/__init__.py b/thirdparty/librdkafka-0.11.1/packaging/nuget/zfile/__init__.py deleted file mode 100644 index e69de29..0000000 http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7528d23e/thirdparty/librdkafka-0.11.1/packaging/nuget/zfile/zfile.py ---------------------------------------------------------------------- diff --git a/thirdparty/librdkafka-0.11.1/packaging/nuget/zfile/zfile.py b/thirdparty/librdkafka-0.11.1/packaging/nuget/zfile/zfile.py deleted file mode 100644 index 8616078..0000000 --- a/thirdparty/librdkafka-0.11.1/packaging/nuget/zfile/zfile.py +++ /dev/null @@ -1,100 +0,0 @@ -#!/usr/bin/env python - -import os -import tarfile -import zipfile -import rpmfile - -class ZFile (object): - def __init__(self, path, mode='r', ext=None): - super(ZFile, self).__init__() - - if ext is not None: - _ext = ext - else: - _ext = os.path.splitext(path)[-1] - if _ext.startswith('.'): - _ext = _ext[1:] - - if zipfile.is_zipfile(path) or _ext == 'zip': - self.f = zipfile.ZipFile(path, mode) - elif tarfile.is_tarfile(path) or _ext in ('tar', 'tgz', 'gz'): - self.f = tarfile.open(path, mode) - elif _ext == 'rpm': - self.f = rpmfile.open(path, mode + 'b') - else: - raise ValueError('Unsupported file extension: %s' % path) - - def __enter__(self): - return self - - def __exit__(self, *args): - if callable(getattr(self.f, 'close', None)): - self.f.close() - - def getnames(self): - if isinstance(self.f, zipfile.ZipFile): - return self.f.namelist() - elif isinstance(self.f, tarfile.TarFile): - return self.f.getnames() - elif isinstance(self.f, rpmfile.RPMFile): - return [x.name for x in self.f.getmembers()] - else: - raise NotImplementedError - - def headers(self): - if isinstance(self.f, rpmfile.RPMFile): - return self.f.headers - else: - return dict() - - def extract_to(self, member, path): - """ Extract compress file's \p member to \p path - If \p path is a directory the member's basename will used as - filename, otherwise path is considered the full file path name. """ - - if not os.path.isdir(os.path.dirname(path)): - os.makedirs(os.path.dirname(path)) - - if os.path.isdir(path): - path = os.path.join(path, os.path.basename(member)) - - with open(path, 'wb') as of: - if isinstance(self.f, zipfile.ZipFile): - zf = self.f.open(member) - else: - zf = self.f.extractfile(member) - - while True: - b = zf.read(1024*100) - if b: - of.write(b) - else: - break - - zf.close() - - - @classmethod - def extract (cls, zpath, member, outpath): - """ - Extract file member (full internal path) to output from - archive zpath. - """ - - with ZFile(zpath) as zf: - zf.extract_to(member, outpath) - - - @classmethod - def compress (cls, zpath, paths, stripcnt=0, ext=None): - """ - Create new compressed file \p zpath containing files in \p paths - """ - - with ZFile(zpath, 'w', ext=ext) as zf: - for p in paths: - outp = os.path.sep.join(p.split(os.path.sep)[stripcnt:]) - print('zip %s to %s (stripcnt %d)' % (p, outp, stripcnt)) - zf.f.write(p, outp) - http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7528d23e/thirdparty/librdkafka-0.11.1/packaging/rpm/.gitignore ---------------------------------------------------------------------- diff --git a/thirdparty/librdkafka-0.11.1/packaging/rpm/.gitignore b/thirdparty/librdkafka-0.11.1/packaging/rpm/.gitignore deleted file mode 100644 index cf122d0..0000000 --- a/thirdparty/librdkafka-0.11.1/packaging/rpm/.gitignore +++ /dev/null @@ -1,3 +0,0 @@ -*.log -available_pkgs -installed_pkgs http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7528d23e/thirdparty/librdkafka-0.11.1/packaging/rpm/Makefile ---------------------------------------------------------------------- diff --git a/thirdparty/librdkafka-0.11.1/packaging/rpm/Makefile b/thirdparty/librdkafka-0.11.1/packaging/rpm/Makefile deleted file mode 100644 index 5e949ca..0000000 --- a/thirdparty/librdkafka-0.11.1/packaging/rpm/Makefile +++ /dev/null @@ -1,79 +0,0 @@ -PACKAGE_NAME?= librdkafka -VERSION?= $(shell ../get_version.py ../../src/rdkafka.h) - -# Jenkins CI integration -BUILD_NUMBER?= 1 - -MOCK_CONFIG?=default - -RESULT_DIR?=pkgs-$(VERSION)-$(BUILD_NUMBER)-$(MOCK_CONFIG) - -all: rpm - - -SOURCES: - mkdir -p SOURCES - -archive: SOURCES - cd ../../ && \ - git archive --prefix=$(PACKAGE_NAME)-$(VERSION)/ \ - -o packaging/rpm/SOURCES/$(PACKAGE_NAME)-$(VERSION).tar.gz HEAD - - -build_prepare: archive - mkdir -p $(RESULT_DIR) - rm -f $(RESULT_DIR)/$(PACKAGE_NAME)*.rpm - - -srpm: build_prepare - /usr/bin/mock \ - -r $(MOCK_CONFIG) \ - --define "__version $(VERSION)" \ - --define "__release $(BUILD_NUMBER)" \ - --resultdir=$(RESULT_DIR) \ - --no-clean --no-cleanup-after \ - --buildsrpm \ - --spec=librdkafka.spec \ - --sources=SOURCES - @echo "======= Source RPM now available in $(RESULT_DIR) =======" - -rpm: srpm - /usr/bin/mock \ - -r $(MOCK_CONFIG) \ - --define "__version $(VERSION)"\ - --define "__release $(BUILD_NUMBER)"\ - --resultdir=$(RESULT_DIR) \ - --no-clean --no-cleanup-after \ - --rebuild $(RESULT_DIR)/$(PACKAGE_NAME)*.src.rpm - @echo "======= Binary RPMs now available in $(RESULT_DIR) =======" - -copy-artifacts: - cp $(RESULT_DIR)/*rpm ../../artifacts/ - -clean: - rm -rf SOURCES - /usr/bin/mock -r $(MOCK_CONFIG) --clean - -distclean: clean - rm -f build.log root.log state.log available_pkgs installed_pkgs \ - *.rpm *.tar.gz - -# Prepare ubuntu 14.04 for building RPMs with mock. -# - older versions of mock needs the config file to reside in /etc/mock, -# so we copy it there. -# - add a mock system group (if not already exists) -# - add the current user to the mock group. -# - prepare mock environment with some needed packages. -# NOTE: This target should be run with sudo. -prepare_ubuntu: - apt-get -qq update - apt-get install -y -qq mock make git python-lzma - cp *.cfg /etc/mock/ - addgroup --system mock || true - adduser $$(whoami) mock - /usr/bin/mock -r $(MOCK_CONFIG) --init - /usr/bin/mock -r $(MOCK_CONFIG) --no-cleanup-after --install epel-release shadow-utils - -prepare_centos: - yum install -y -q mock make git - cp *.cfg /etc/mock/ http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7528d23e/thirdparty/librdkafka-0.11.1/packaging/rpm/el7-x86_64.cfg ---------------------------------------------------------------------- diff --git a/thirdparty/librdkafka-0.11.1/packaging/rpm/el7-x86_64.cfg b/thirdparty/librdkafka-0.11.1/packaging/rpm/el7-x86_64.cfg deleted file mode 100644 index 5022827..0000000 --- a/thirdparty/librdkafka-0.11.1/packaging/rpm/el7-x86_64.cfg +++ /dev/null @@ -1,40 +0,0 @@ -config_opts['root'] = 'el7-x86_64' -config_opts['target_arch'] = 'x86_64' -config_opts['legal_host_arches'] = ('x86_64',) -config_opts['chroot_setup_cmd'] = 'install @buildsys-build' -config_opts['dist'] = 'el7' # only useful for --resultdir variable subst -config_opts['releasever'] = '7' -config_opts['docker_unshare_warning'] = False -config_opts['nosync'] = True - -config_opts['yum.conf'] = """ -[main] -keepcache=1 -debuglevel=2 -reposdir=/dev/null -logfile=/var/log/yum.log -retries=15 -obsoletes=1 -gpgcheck=0 -assumeyes=1 -syslog_ident=mock -syslog_device= -mdpolicy=group:primary - -# repos -[base] -name=BaseOS -mirrorlist=http://mirrorlist.centos.org/?release=7&arch=x86_64&repo=os -failovermethod=priority - -[updates] -name=updates -enabled=1 -mirrorlist=http://mirrorlist.centos.org/?release=7&arch=x86_64&repo=updates -failovermethod=priority - -[epel] -name=epel -mirrorlist=http://mirrors.fedoraproject.org/mirrorlist?repo=epel-7&arch=x86_64 -failovermethod=priority -""" http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7528d23e/thirdparty/librdkafka-0.11.1/packaging/rpm/librdkafka.spec ---------------------------------------------------------------------- diff --git a/thirdparty/librdkafka-0.11.1/packaging/rpm/librdkafka.spec b/thirdparty/librdkafka-0.11.1/packaging/rpm/librdkafka.spec deleted file mode 100644 index 8027acf..0000000 --- a/thirdparty/librdkafka-0.11.1/packaging/rpm/librdkafka.spec +++ /dev/null @@ -1,103 +0,0 @@ -Name: librdkafka -Version: %{__version} -Release: %{__release}%{?dist} -%define soname 1 - -Summary: The Apache Kafka C library -Group: Development/Libraries/C and C++ -License: BSD-2-Clause -URL: https://github.com/edenhill/librdkafka -Source: librdkafka-%{version}.tar.gz - -BuildRequires: zlib-devel libstdc++-devel gcc >= 4.1 gcc-c++ openssl-devel cyrus-sasl-devel lz4-devel python -BuildRoot: %(mktemp -ud %{_tmppath}/%{name}-%{version}-%{release}-XXXXXX) - -%define _source_payload w9.gzdio -%define _binary_payload w9.gzdio - -%description -librdkafka is the C/C++ client library implementation of the Apache Kafka protocol, containing both Producer and Consumer support. - - -%package -n %{name}%{soname} -Summary: The Apache Kafka C library -Group: Development/Libraries/C and C++ -Requires: zlib libstdc++ cyrus-sasl -# openssl libraries were extract to openssl-libs in RHEL7 -%if 0%{?rhel} >= 7 -Requires: openssl-libs -%else -Requires: openssl -%endif - -%description -n %{name}%{soname} -librdkafka is the C/C++ client library implementation of the Apache Kafka protocol, containing both Producer and Consumer support. - - -%package -n %{name}-devel -Summary: The Apache Kafka C library (Development Environment) -Group: Development/Libraries/C and C++ -Requires: %{name}%{soname} = %{version} - -%description -n %{name}-devel -librdkafka is the C/C++ client library implementation of the Apache Kafka protocol, containing both Producer and Consumer support. - -This package contains headers and libraries required to build applications -using librdkafka. - - -%prep -%setup -q -n %{name}-%{version} - -%configure - -%build -make - -%install -rm -rf %{buildroot} -DESTDIR=%{buildroot} make install - -%clean -rm -rf %{buildroot} - -%post -n %{name}%{soname} -p /sbin/ldconfig -%postun -n %{name}%{soname} -p /sbin/ldconfig - -%files -n %{name}%{soname} -%defattr(444,root,root) -%{_libdir}/librdkafka.so.%{soname} -%{_libdir}/librdkafka++.so.%{soname} -%defattr(-,root,root) -%doc README.md CONFIGURATION.md INTRODUCTION.md -%doc LICENSE LICENSE.pycrc LICENSE.queue LICENSE.snappy LICENSE.tinycthread LICENSE.wingetopt - -%defattr(-,root,root) -#%{_bindir}/rdkafka_example -#%{_bindir}/rdkafka_performance - - -%files -n %{name}-devel -%defattr(-,root,root) -%{_includedir}/librdkafka -%defattr(444,root,root) -%{_libdir}/librdkafka.a -%{_libdir}/librdkafka.so -%{_libdir}/librdkafka++.a -%{_libdir}/librdkafka++.so -%{_libdir}/pkgconfig/rdkafka++.pc -%{_libdir}/pkgconfig/rdkafka.pc - - -%changelog -* Thu Apr 09 2015 Eduard Iskandarov <[email protected]> 0.8.6-0 -- 0.8.6 simplify build process - -* Fri Oct 24 2014 Magnus Edenhill <[email protected]> 0.8.5-0 -- 0.8.5 release - -* Mon Aug 18 2014 Magnus Edenhill <[email protected]> 0.8.4-0 -- 0.8.4 release - -* Mon Mar 17 2014 Magnus Edenhill <[email protected]> 0.8.3-0 -- Initial RPM package http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7528d23e/thirdparty/librdkafka-0.11.1/src-cpp/CMakeLists.txt ---------------------------------------------------------------------- diff --git a/thirdparty/librdkafka-0.11.1/src-cpp/CMakeLists.txt b/thirdparty/librdkafka-0.11.1/src-cpp/CMakeLists.txt deleted file mode 100644 index bcbc4ae..0000000 --- a/thirdparty/librdkafka-0.11.1/src-cpp/CMakeLists.txt +++ /dev/null @@ -1,33 +0,0 @@ -add_library( - rdkafka++ - ConfImpl.cpp - ConsumerImpl.cpp - HandleImpl.cpp - KafkaConsumerImpl.cpp - MessageImpl.cpp - MetadataImpl.cpp - ProducerImpl.cpp - QueueImpl.cpp - RdKafka.cpp - TopicImpl.cpp - TopicPartitionImpl.cpp -) - -target_link_libraries(rdkafka++ PUBLIC rdkafka) - -# Support '#include <rdkafcpp.h>' -target_include_directories(rdkafka++ PUBLIC "$<BUILD_INTERFACE:${CMAKE_CURRENT_LIST_DIR}>") - -install( - TARGETS rdkafka++ - EXPORT "${targets_export_name}" - LIBRARY DESTINATION "${CMAKE_INSTALL_LIBDIR}" - ARCHIVE DESTINATION "${CMAKE_INSTALL_LIBDIR}" - RUNTIME DESTINATION "${CMAKE_INSTALL_BINDIR}" - INCLUDES DESTINATION "${CMAKE_INSTALL_INCLUDEDIR}" -) - -install( - FILES "rdkafkacpp.h" - DESTINATION "${CMAKE_INSTALL_INCLUDEDIR}/librdkafka" -) http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7528d23e/thirdparty/librdkafka-0.11.1/src-cpp/ConfImpl.cpp ---------------------------------------------------------------------- diff --git a/thirdparty/librdkafka-0.11.1/src-cpp/ConfImpl.cpp b/thirdparty/librdkafka-0.11.1/src-cpp/ConfImpl.cpp deleted file mode 100644 index 709c728..0000000 --- a/thirdparty/librdkafka-0.11.1/src-cpp/ConfImpl.cpp +++ /dev/null @@ -1,89 +0,0 @@ -/* - * librdkafka - Apache Kafka C/C++ library - * - * Copyright (c) 2014 Magnus Edenhill - * All rights reserved. - * - * Redistribution and use in source and binary forms, with or without - * modification, are permitted provided that the following conditions are met: - * - * 1. Redistributions of source code must retain the above copyright notice, - * this list of conditions and the following disclaimer. - * 2. Redistributions in binary form must reproduce the above copyright notice, - * this list of conditions and the following disclaimer in the documentation - * and/or other materials provided with the distribution. - * - * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" - * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE - * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE - * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE - * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR - * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF - * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS - * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN - * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) - * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE - * POSSIBILITY OF SUCH DAMAGE. - */ - -#include <iostream> -#include <string> -#include <list> - -#include "rdkafkacpp_int.h" - - - -RdKafka::ConfImpl::ConfResult RdKafka::ConfImpl::set(const std::string &name, - const std::string &value, - std::string &errstr) { - rd_kafka_conf_res_t res; - char errbuf[512]; - - if (this->conf_type_ == CONF_GLOBAL) - res = rd_kafka_conf_set(this->rk_conf_, - name.c_str(), value.c_str(), - errbuf, sizeof(errbuf)); - else - res = rd_kafka_topic_conf_set(this->rkt_conf_, - name.c_str(), value.c_str(), - errbuf, sizeof(errbuf)); - - if (res != RD_KAFKA_CONF_OK) - errstr = errbuf; - - return static_cast<Conf::ConfResult>(res); -} - - -std::list<std::string> *RdKafka::ConfImpl::dump () { - - const char **arrc; - size_t cnt; - std::list<std::string> *arr; - - if (rk_conf_) - arrc = rd_kafka_conf_dump(rk_conf_, &cnt); - else - arrc = rd_kafka_topic_conf_dump(rkt_conf_, &cnt); - - arr = new std::list<std::string>(); - for (int i = 0 ; i < static_cast<int>(cnt) ; i++) - arr->push_back(std::string(arrc[i])); - - rd_kafka_conf_dump_free(arrc, cnt); - return arr; -} - -RdKafka::Conf *RdKafka::Conf::create (ConfType type) { - ConfImpl *conf = new ConfImpl(); - - conf->conf_type_ = type; - - if (type == CONF_GLOBAL) - conf->rk_conf_ = rd_kafka_conf_new(); - else - conf->rkt_conf_ = rd_kafka_topic_conf_new(); - - return conf; -} http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7528d23e/thirdparty/librdkafka-0.11.1/src-cpp/ConsumerImpl.cpp ---------------------------------------------------------------------- diff --git a/thirdparty/librdkafka-0.11.1/src-cpp/ConsumerImpl.cpp b/thirdparty/librdkafka-0.11.1/src-cpp/ConsumerImpl.cpp deleted file mode 100644 index bb46877..0000000 --- a/thirdparty/librdkafka-0.11.1/src-cpp/ConsumerImpl.cpp +++ /dev/null @@ -1,233 +0,0 @@ -/* - * librdkafka - Apache Kafka C/C++ library - * - * Copyright (c) 2014 Magnus Edenhill - * All rights reserved. - * - * Redistribution and use in source and binary forms, with or without - * modification, are permitted provided that the following conditions are met: - * - * 1. Redistributions of source code must retain the above copyright notice, - * this list of conditions and the following disclaimer. - * 2. Redistributions in binary form must reproduce the above copyright notice, - * this list of conditions and the following disclaimer in the documentation - * and/or other materials provided with the distribution. - * - * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" - * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE - * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE - * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE - * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR - * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF - * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS - * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN - * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) - * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE - * POSSIBILITY OF SUCH DAMAGE. - */ - -#include <iostream> -#include <string> -#include <list> -#include <cerrno> - -#include "rdkafkacpp_int.h" - -RdKafka::Consumer::~Consumer () {} - -RdKafka::Consumer *RdKafka::Consumer::create (RdKafka::Conf *conf, - std::string &errstr) { - char errbuf[512]; - RdKafka::ConfImpl *confimpl = dynamic_cast<RdKafka::ConfImpl *>(conf); - RdKafka::ConsumerImpl *rkc = new RdKafka::ConsumerImpl(); - rd_kafka_conf_t *rk_conf = NULL; - - if (confimpl) { - if (!confimpl->rk_conf_) { - errstr = "Requires RdKafka::Conf::CONF_GLOBAL object"; - delete rkc; - return NULL; - } - - rkc->set_common_config(confimpl); - - rk_conf = rd_kafka_conf_dup(confimpl->rk_conf_); - } - - rd_kafka_t *rk; - if (!(rk = rd_kafka_new(RD_KAFKA_CONSUMER, rk_conf, - errbuf, sizeof(errbuf)))) { - errstr = errbuf; - delete rkc; - return NULL; - } - - rkc->rk_ = rk; - - - return rkc; -} - -int64_t RdKafka::Consumer::OffsetTail (int64_t offset) { - return RD_KAFKA_OFFSET_TAIL(offset); -} - -RdKafka::ErrorCode RdKafka::ConsumerImpl::start (Topic *topic, - int32_t partition, - int64_t offset) { - RdKafka::TopicImpl *topicimpl = dynamic_cast<RdKafka::TopicImpl *>(topic); - - if (rd_kafka_consume_start(topicimpl->rkt_, partition, offset) == -1) - return static_cast<RdKafka::ErrorCode>(rd_kafka_last_error()); - - return RdKafka::ERR_NO_ERROR; -} - - -RdKafka::ErrorCode RdKafka::ConsumerImpl::start (Topic *topic, - int32_t partition, - int64_t offset, - Queue *queue) { - RdKafka::TopicImpl *topicimpl = dynamic_cast<RdKafka::TopicImpl *>(topic); - RdKafka::QueueImpl *queueimpl = dynamic_cast<RdKafka::QueueImpl *>(queue); - - if (rd_kafka_consume_start_queue(topicimpl->rkt_, partition, offset, - queueimpl->queue_) == -1) - return static_cast<RdKafka::ErrorCode>(rd_kafka_last_error()); - - return RdKafka::ERR_NO_ERROR; -} - - -RdKafka::ErrorCode RdKafka::ConsumerImpl::stop (Topic *topic, - int32_t partition) { - RdKafka::TopicImpl *topicimpl = dynamic_cast<RdKafka::TopicImpl *>(topic); - - if (rd_kafka_consume_stop(topicimpl->rkt_, partition) == -1) - return static_cast<RdKafka::ErrorCode>(rd_kafka_last_error()); - - return RdKafka::ERR_NO_ERROR; -} - -RdKafka::ErrorCode RdKafka::ConsumerImpl::seek (Topic *topic, - int32_t partition, - int64_t offset, - int timeout_ms) { - RdKafka::TopicImpl *topicimpl = dynamic_cast<RdKafka::TopicImpl *>(topic); - - if (rd_kafka_seek(topicimpl->rkt_, partition, offset, timeout_ms) == -1) - return static_cast<RdKafka::ErrorCode>(rd_kafka_last_error()); - - return RdKafka::ERR_NO_ERROR; -} - -RdKafka::Message *RdKafka::ConsumerImpl::consume (Topic *topic, - int32_t partition, - int timeout_ms) { - RdKafka::TopicImpl *topicimpl = dynamic_cast<RdKafka::TopicImpl *>(topic); - rd_kafka_message_t *rkmessage; - - rkmessage = rd_kafka_consume(topicimpl->rkt_, partition, timeout_ms); - if (!rkmessage) - return new RdKafka::MessageImpl(topic, - static_cast<RdKafka::ErrorCode> - (rd_kafka_last_error())); - - return new RdKafka::MessageImpl(topic, rkmessage); -} - -namespace { - /* Helper struct for `consume_callback'. - * Encapsulates the values we need in order to call `rd_kafka_consume_callback' - * and keep track of the C++ callback function and `opaque' value. - */ - struct ConsumerImplCallback { - ConsumerImplCallback(RdKafka::Topic* topic, RdKafka::ConsumeCb* cb, void* data) - : topic(topic), cb_cls(cb), cb_data(data) { - } - /* This function is the one we give to `rd_kafka_consume_callback', with - * the `opaque' pointer pointing to an instance of this struct, in which - * we can find the C++ callback and `cb_data'. - */ - static void consume_cb_trampoline(rd_kafka_message_t *msg, void *opaque) { - ConsumerImplCallback *instance = static_cast<ConsumerImplCallback*>(opaque); - RdKafka::MessageImpl message(instance->topic, msg, false /*don't free*/); - instance->cb_cls->consume_cb(message, instance->cb_data); - } - RdKafka::Topic *topic; - RdKafka::ConsumeCb *cb_cls; - void *cb_data; - }; -} - -int RdKafka::ConsumerImpl::consume_callback (RdKafka::Topic* topic, - int32_t partition, - int timeout_ms, - RdKafka::ConsumeCb *consume_cb, - void *opaque) { - RdKafka::TopicImpl *topicimpl = static_cast<RdKafka::TopicImpl *>(topic); - ConsumerImplCallback context(topic, consume_cb, opaque); - return rd_kafka_consume_callback(topicimpl->rkt_, partition, timeout_ms, - &ConsumerImplCallback::consume_cb_trampoline, &context); -} - - -RdKafka::Message *RdKafka::ConsumerImpl::consume (Queue *queue, - int timeout_ms) { - RdKafka::QueueImpl *queueimpl = dynamic_cast<RdKafka::QueueImpl *>(queue); - rd_kafka_message_t *rkmessage; - - rkmessage = rd_kafka_consume_queue(queueimpl->queue_, timeout_ms); - if (!rkmessage) - return new RdKafka::MessageImpl(NULL, - static_cast<RdKafka::ErrorCode> - (rd_kafka_last_error())); - /* - * Recover our Topic * from the topic conf's opaque field, which we - * set in RdKafka::Topic::create() for just this kind of situation. - */ - void *opaque = rd_kafka_topic_opaque(rkmessage->rkt); - Topic *topic = static_cast<Topic *>(opaque); - - return new RdKafka::MessageImpl(topic, rkmessage); -} - -namespace { - /* Helper struct for `consume_callback' with a Queue. - * Encapsulates the values we need in order to call `rd_kafka_consume_callback' - * and keep track of the C++ callback function and `opaque' value. - */ - struct ConsumerImplQueueCallback { - ConsumerImplQueueCallback(RdKafka::ConsumeCb *cb, void *data) - : cb_cls(cb), cb_data(data) { - } - /* This function is the one we give to `rd_kafka_consume_callback', with - * the `opaque' pointer pointing to an instance of this struct, in which - * we can find the C++ callback and `cb_data'. - */ - static void consume_cb_trampoline(rd_kafka_message_t *msg, void *opaque) { - ConsumerImplQueueCallback *instance = static_cast<ConsumerImplQueueCallback *>(opaque); - /* - * Recover our Topic * from the topic conf's opaque field, which we - * set in RdKafka::Topic::create() for just this kind of situation. - */ - void *topic_opaque = rd_kafka_topic_opaque(msg->rkt); - RdKafka::Topic *topic = static_cast<RdKafka::Topic *>(topic_opaque); - RdKafka::MessageImpl message(topic, msg, false /*don't free*/); - instance->cb_cls->consume_cb(message, instance->cb_data); - } - RdKafka::ConsumeCb *cb_cls; - void *cb_data; - }; -} - -int RdKafka::ConsumerImpl::consume_callback (Queue *queue, - int timeout_ms, - RdKafka::ConsumeCb *consume_cb, - void *opaque) { - RdKafka::QueueImpl *queueimpl = dynamic_cast<RdKafka::QueueImpl *>(queue); - ConsumerImplQueueCallback context(consume_cb, opaque); - return rd_kafka_consume_callback_queue(queueimpl->queue_, timeout_ms, - &ConsumerImplQueueCallback::consume_cb_trampoline, - &context); -} http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7528d23e/thirdparty/librdkafka-0.11.1/src-cpp/HandleImpl.cpp ---------------------------------------------------------------------- diff --git a/thirdparty/librdkafka-0.11.1/src-cpp/HandleImpl.cpp b/thirdparty/librdkafka-0.11.1/src-cpp/HandleImpl.cpp deleted file mode 100644 index 3bdccbf..0000000 --- a/thirdparty/librdkafka-0.11.1/src-cpp/HandleImpl.cpp +++ /dev/null @@ -1,365 +0,0 @@ -/* - * librdkafka - Apache Kafka C/C++ library - * - * Copyright (c) 2014 Magnus Edenhill - * All rights reserved. - * - * Redistribution and use in source and binary forms, with or without - * modification, are permitted provided that the following conditions are met: - * - * 1. Redistributions of source code must retain the above copyright notice, - * this list of conditions and the following disclaimer. - * 2. Redistributions in binary form must reproduce the above copyright notice, - * this list of conditions and the following disclaimer in the documentation - * and/or other materials provided with the distribution. - * - * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" - * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE - * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE - * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE - * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR - * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF - * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS - * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN - * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) - * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE - * POSSIBILITY OF SUCH DAMAGE. - */ - -#include <iostream> -#include <string> -#include <list> - -#include "rdkafkacpp_int.h" - -void RdKafka::consume_cb_trampoline(rd_kafka_message_t *msg, void *opaque) { - RdKafka::HandleImpl *handle = static_cast<RdKafka::HandleImpl *>(opaque); - RdKafka::Topic* topic = static_cast<Topic *>(rd_kafka_topic_opaque(msg->rkt)); - - RdKafka::MessageImpl message(topic, msg, false /*don't free*/); - - handle->consume_cb_->consume_cb(message, opaque); -} - -void RdKafka::log_cb_trampoline (const rd_kafka_t *rk, int level, - const char *fac, const char *buf) { - if (!rk) { - rd_kafka_log_print(rk, level, fac, buf); - return; - } - - void *opaque = rd_kafka_opaque(rk); - RdKafka::HandleImpl *handle = static_cast<RdKafka::HandleImpl *>(opaque); - - if (!handle->event_cb_) { - rd_kafka_log_print(rk, level, fac, buf); - return; - } - - RdKafka::EventImpl event(RdKafka::Event::EVENT_LOG, - RdKafka::ERR_NO_ERROR, - static_cast<RdKafka::Event::Severity>(level), - fac, buf); - - handle->event_cb_->event_cb(event); -} - - -void RdKafka::error_cb_trampoline (rd_kafka_t *rk, int err, - const char *reason, void *opaque) { - RdKafka::HandleImpl *handle = static_cast<RdKafka::HandleImpl *>(opaque); - - RdKafka::EventImpl event(RdKafka::Event::EVENT_ERROR, - static_cast<RdKafka::ErrorCode>(err), - RdKafka::Event::EVENT_SEVERITY_ERROR, - NULL, - reason); - - handle->event_cb_->event_cb(event); -} - - -void RdKafka::throttle_cb_trampoline (rd_kafka_t *rk, const char *broker_name, - int32_t broker_id, - int throttle_time_ms, - void *opaque) { - RdKafka::HandleImpl *handle = static_cast<RdKafka::HandleImpl *>(opaque); - - RdKafka::EventImpl event(RdKafka::Event::EVENT_THROTTLE); - event.str_ = broker_name; - event.id_ = broker_id; - event.throttle_time_ = throttle_time_ms; - - handle->event_cb_->event_cb(event); -} - - -int RdKafka::stats_cb_trampoline (rd_kafka_t *rk, char *json, size_t json_len, - void *opaque) { - RdKafka::HandleImpl *handle = static_cast<RdKafka::HandleImpl *>(opaque); - - RdKafka::EventImpl event(RdKafka::Event::EVENT_STATS, - RdKafka::ERR_NO_ERROR, - RdKafka::Event::EVENT_SEVERITY_INFO, - NULL, json); - - handle->event_cb_->event_cb(event); - - return 0; -} - - -int RdKafka::socket_cb_trampoline (int domain, int type, int protocol, - void *opaque) { - RdKafka::HandleImpl *handle = static_cast<RdKafka::HandleImpl *>(opaque); - - return handle->socket_cb_->socket_cb(domain, type, protocol); -} - -int RdKafka::open_cb_trampoline (const char *pathname, int flags, mode_t mode, - void *opaque) { - RdKafka::HandleImpl *handle = static_cast<RdKafka::HandleImpl *>(opaque); - - return handle->open_cb_->open_cb(pathname, flags, static_cast<int>(mode)); -} - -RdKafka::ErrorCode RdKafka::HandleImpl::metadata (bool all_topics, - const Topic *only_rkt, - Metadata **metadatap, - int timeout_ms) { - - const rd_kafka_metadata_t *cmetadatap=NULL; - - rd_kafka_topic_t *topic = only_rkt ? - static_cast<const TopicImpl *>(only_rkt)->rkt_ : NULL; - - const rd_kafka_resp_err_t rc = rd_kafka_metadata(rk_, all_topics, topic, - &cmetadatap,timeout_ms); - - *metadatap = (rc == RD_KAFKA_RESP_ERR_NO_ERROR) ? - new RdKafka::MetadataImpl(cmetadatap) : NULL; - - return static_cast<RdKafka::ErrorCode>(rc); -} - -/** - * Convert a list of C partitions to C++ partitions - */ -static void c_parts_to_partitions (const rd_kafka_topic_partition_list_t - *c_parts, - std::vector<RdKafka::TopicPartition*> - &partitions) { - partitions.resize(c_parts->cnt); - for (int i = 0 ; i < c_parts->cnt ; i++) - partitions[i] = new RdKafka::TopicPartitionImpl(&c_parts->elems[i]); -} - -static void free_partition_vector (std::vector<RdKafka::TopicPartition*> &v) { - for (unsigned int i = 0 ; i < v.size() ; i++) - delete v[i]; - v.clear(); -} - -void -RdKafka::rebalance_cb_trampoline (rd_kafka_t *rk, - rd_kafka_resp_err_t err, - rd_kafka_topic_partition_list_t *c_partitions, - void *opaque) { - RdKafka::HandleImpl *handle = static_cast<RdKafka::HandleImpl *>(opaque); - std::vector<RdKafka::TopicPartition*> partitions; - - c_parts_to_partitions(c_partitions, partitions); - - handle->rebalance_cb_->rebalance_cb( - dynamic_cast<RdKafka::KafkaConsumer*>(handle), - static_cast<RdKafka::ErrorCode>(err), - partitions); - - free_partition_vector(partitions); -} - - -void -RdKafka::offset_commit_cb_trampoline0 ( - rd_kafka_t *rk, - rd_kafka_resp_err_t err, - rd_kafka_topic_partition_list_t *c_offsets, void *opaque) { - OffsetCommitCb *cb = static_cast<RdKafka::OffsetCommitCb *>(opaque); - std::vector<RdKafka::TopicPartition*> offsets; - - if (c_offsets) - c_parts_to_partitions(c_offsets, offsets); - - cb->offset_commit_cb(static_cast<RdKafka::ErrorCode>(err), offsets); - - free_partition_vector(offsets); -} - -static void -offset_commit_cb_trampoline ( - rd_kafka_t *rk, - rd_kafka_resp_err_t err, - rd_kafka_topic_partition_list_t *c_offsets, void *opaque) { - RdKafka::HandleImpl *handle = static_cast<RdKafka::HandleImpl *>(opaque); - RdKafka::offset_commit_cb_trampoline0(rk, err, c_offsets, - handle->offset_commit_cb_); -} - - -void RdKafka::HandleImpl::set_common_config (RdKafka::ConfImpl *confimpl) { - - rd_kafka_conf_set_opaque(confimpl->rk_conf_, this); - - if (confimpl->event_cb_) { - rd_kafka_conf_set_log_cb(confimpl->rk_conf_, - RdKafka::log_cb_trampoline); - rd_kafka_conf_set_error_cb(confimpl->rk_conf_, - RdKafka::error_cb_trampoline); - rd_kafka_conf_set_throttle_cb(confimpl->rk_conf_, - RdKafka::throttle_cb_trampoline); - rd_kafka_conf_set_stats_cb(confimpl->rk_conf_, - RdKafka::stats_cb_trampoline); - event_cb_ = confimpl->event_cb_; - } - - if (confimpl->socket_cb_) { - rd_kafka_conf_set_socket_cb(confimpl->rk_conf_, - RdKafka::socket_cb_trampoline); - socket_cb_ = confimpl->socket_cb_; - } - - if (confimpl->open_cb_) { -#ifndef _MSC_VER - rd_kafka_conf_set_open_cb(confimpl->rk_conf_, RdKafka::open_cb_trampoline); - open_cb_ = confimpl->open_cb_; -#endif - } - - if (confimpl->rebalance_cb_) { - rd_kafka_conf_set_rebalance_cb(confimpl->rk_conf_, - RdKafka::rebalance_cb_trampoline); - rebalance_cb_ = confimpl->rebalance_cb_; - } - - if (confimpl->offset_commit_cb_) { - rd_kafka_conf_set_offset_commit_cb(confimpl->rk_conf_, - offset_commit_cb_trampoline); - offset_commit_cb_ = confimpl->offset_commit_cb_; - } - - if (confimpl->consume_cb_) { - rd_kafka_conf_set_consume_cb(confimpl->rk_conf_, - RdKafka::consume_cb_trampoline); - consume_cb_ = confimpl->consume_cb_; - } - -} - - -RdKafka::ErrorCode -RdKafka::HandleImpl::pause (std::vector<RdKafka::TopicPartition*> &partitions) { - rd_kafka_topic_partition_list_t *c_parts; - rd_kafka_resp_err_t err; - - c_parts = partitions_to_c_parts(partitions); - - err = rd_kafka_pause_partitions(rk_, c_parts); - - if (!err) - update_partitions_from_c_parts(partitions, c_parts); - - rd_kafka_topic_partition_list_destroy(c_parts); - - return static_cast<RdKafka::ErrorCode>(err); -} - - -RdKafka::ErrorCode -RdKafka::HandleImpl::resume (std::vector<RdKafka::TopicPartition*> &partitions) { - rd_kafka_topic_partition_list_t *c_parts; - rd_kafka_resp_err_t err; - - c_parts = partitions_to_c_parts(partitions); - - err = rd_kafka_resume_partitions(rk_, c_parts); - - if (!err) - update_partitions_from_c_parts(partitions, c_parts); - - rd_kafka_topic_partition_list_destroy(c_parts); - - return static_cast<RdKafka::ErrorCode>(err); -} - -RdKafka::Queue * -RdKafka::HandleImpl::get_partition_queue (const TopicPartition *part) { - rd_kafka_queue_t *rkqu; - rkqu = rd_kafka_queue_get_partition(rk_, - part->topic().c_str(), - part->partition()); - - if (rkqu == NULL) - return NULL; - - RdKafka::QueueImpl *queueimpl = new RdKafka::QueueImpl; - queueimpl->queue_ = rkqu; - - return queueimpl; -} - -RdKafka::ErrorCode -RdKafka::HandleImpl::set_log_queue (RdKafka::Queue *queue) { - rd_kafka_queue_t *rkqu = NULL; - if (queue) { - QueueImpl *queueimpl = dynamic_cast<QueueImpl *>(queue); - rkqu = queueimpl->queue_; - } - return static_cast<RdKafka::ErrorCode>( - rd_kafka_set_log_queue(rk_, rkqu)); -} - -namespace RdKafka { - -rd_kafka_topic_partition_list_t * -partitions_to_c_parts (const std::vector<RdKafka::TopicPartition*> &partitions){ - rd_kafka_topic_partition_list_t *c_parts; - - c_parts = rd_kafka_topic_partition_list_new((int)partitions.size()); - - for (unsigned int i = 0 ; i < partitions.size() ; i++) { - const RdKafka::TopicPartitionImpl *tpi = - dynamic_cast<const RdKafka::TopicPartitionImpl*>(partitions[i]); - rd_kafka_topic_partition_t *rktpar = - rd_kafka_topic_partition_list_add(c_parts, - tpi->topic_.c_str(), tpi->partition_); - rktpar->offset = tpi->offset_; - } - - return c_parts; -} - - -/** - * @brief Update the application provided 'partitions' with info from 'c_parts' - */ -void -update_partitions_from_c_parts (std::vector<RdKafka::TopicPartition*> &partitions, - const rd_kafka_topic_partition_list_t *c_parts) { - for (int i = 0 ; i < c_parts->cnt ; i++) { - rd_kafka_topic_partition_t *p = &c_parts->elems[i]; - - /* Find corresponding C++ entry */ - for (unsigned int j = 0 ; j < partitions.size() ; j++) { - RdKafka::TopicPartitionImpl *pp = - dynamic_cast<RdKafka::TopicPartitionImpl*>(partitions[j]); - if (!strcmp(p->topic, pp->topic_.c_str()) && - p->partition == pp->partition_) { - pp->offset_ = p->offset; - pp->err_ = static_cast<RdKafka::ErrorCode>(p->err); - } - } - } -} - -}; - http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7528d23e/thirdparty/librdkafka-0.11.1/src-cpp/KafkaConsumerImpl.cpp ---------------------------------------------------------------------- diff --git a/thirdparty/librdkafka-0.11.1/src-cpp/KafkaConsumerImpl.cpp b/thirdparty/librdkafka-0.11.1/src-cpp/KafkaConsumerImpl.cpp deleted file mode 100644 index f4e79d3..0000000 --- a/thirdparty/librdkafka-0.11.1/src-cpp/KafkaConsumerImpl.cpp +++ /dev/null @@ -1,257 +0,0 @@ -/* - * librdkafka - Apache Kafka C/C++ library - * - * Copyright (c) 2015 Magnus Edenhill - * All rights reserved. - * - * Redistribution and use in source and binary forms, with or without - * modification, are permitted provided that the following conditions are met: - * - * 1. Redistributions of source code must retain the above copyright notice, - * this list of conditions and the following disclaimer. - * 2. Redistributions in binary form must reproduce the above copyright notice, - * this list of conditions and the following disclaimer in the documentation - * and/or other materials provided with the distribution. - * - * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" - * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE - * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE - * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE - * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR - * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF - * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS - * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN - * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) - * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE - * POSSIBILITY OF SUCH DAMAGE. - */ - -#include <string> -#include <vector> - -#include "rdkafkacpp_int.h" - -RdKafka::KafkaConsumer::~KafkaConsumer () {} - -RdKafka::KafkaConsumer *RdKafka::KafkaConsumer::create (RdKafka::Conf *conf, - std::string &errstr) { - char errbuf[512]; - RdKafka::ConfImpl *confimpl = dynamic_cast<RdKafka::ConfImpl *>(conf); - RdKafka::KafkaConsumerImpl *rkc = new RdKafka::KafkaConsumerImpl(); - rd_kafka_conf_t *rk_conf = NULL; - size_t grlen; - - if (!confimpl->rk_conf_) { - errstr = "Requires RdKafka::Conf::CONF_GLOBAL object"; - delete rkc; - return NULL; - } - - if (rd_kafka_conf_get(confimpl->rk_conf_, "group.id", - NULL, &grlen) != RD_KAFKA_CONF_OK || - grlen <= 1 /* terminating null only */) { - errstr = "\"group.id\" must be configured"; - delete rkc; - return NULL; - } - - rkc->set_common_config(confimpl); - - rk_conf = rd_kafka_conf_dup(confimpl->rk_conf_); - - rd_kafka_t *rk; - if (!(rk = rd_kafka_new(RD_KAFKA_CONSUMER, rk_conf, - errbuf, sizeof(errbuf)))) { - errstr = errbuf; - delete rkc; - return NULL; - } - - rkc->rk_ = rk; - - /* Redirect handle queue to cgrp's queue to provide a single queue point */ - rd_kafka_poll_set_consumer(rk); - - return rkc; -} - - - - - - - -RdKafka::ErrorCode -RdKafka::KafkaConsumerImpl::subscribe (const std::vector<std::string> &topics) { - rd_kafka_topic_partition_list_t *c_topics; - rd_kafka_resp_err_t err; - - c_topics = rd_kafka_topic_partition_list_new((int)topics.size()); - - for (unsigned int i = 0 ; i < topics.size() ; i++) - rd_kafka_topic_partition_list_add(c_topics, topics[i].c_str(), - RD_KAFKA_PARTITION_UA); - - err = rd_kafka_subscribe(rk_, c_topics); - - rd_kafka_topic_partition_list_destroy(c_topics); - - return static_cast<RdKafka::ErrorCode>(err); -} - - - -RdKafka::ErrorCode -RdKafka::KafkaConsumerImpl::unsubscribe () { - return static_cast<RdKafka::ErrorCode>(rd_kafka_unsubscribe(this->rk_)); -} - -RdKafka::Message *RdKafka::KafkaConsumerImpl::consume (int timeout_ms) { - rd_kafka_message_t *rkmessage; - - rkmessage = rd_kafka_consumer_poll(this->rk_, timeout_ms); - - if (!rkmessage) - return new RdKafka::MessageImpl(NULL, RdKafka::ERR__TIMED_OUT); - - return new RdKafka::MessageImpl(rkmessage); - -} - - - -RdKafka::ErrorCode -RdKafka::KafkaConsumerImpl::assignment (std::vector<RdKafka::TopicPartition*> &partitions) { - rd_kafka_topic_partition_list_t *c_parts; - rd_kafka_resp_err_t err; - - if ((err = rd_kafka_assignment(rk_, &c_parts))) - return static_cast<RdKafka::ErrorCode>(err); - - partitions.resize(c_parts->cnt); - - for (int i = 0 ; i < c_parts->cnt ; i++) - partitions[i] = new RdKafka::TopicPartitionImpl(&c_parts->elems[i]); - - rd_kafka_topic_partition_list_destroy(c_parts); - - return RdKafka::ERR_NO_ERROR; -} - - -RdKafka::ErrorCode -RdKafka::KafkaConsumerImpl::subscription (std::vector<std::string> &topics) { - rd_kafka_topic_partition_list_t *c_topics; - rd_kafka_resp_err_t err; - - if ((err = rd_kafka_subscription(rk_, &c_topics))) - return static_cast<RdKafka::ErrorCode>(err); - - topics.resize(c_topics->cnt); - for (int i = 0 ; i < c_topics->cnt ; i++) - topics[i] = std::string(c_topics->elems[i].topic); - - rd_kafka_topic_partition_list_destroy(c_topics); - - return RdKafka::ERR_NO_ERROR; -} - - -RdKafka::ErrorCode -RdKafka::KafkaConsumerImpl::assign (const std::vector<TopicPartition*> &partitions) { - rd_kafka_topic_partition_list_t *c_parts; - rd_kafka_resp_err_t err; - - c_parts = partitions_to_c_parts(partitions); - - err = rd_kafka_assign(rk_, c_parts); - - rd_kafka_topic_partition_list_destroy(c_parts); - return static_cast<RdKafka::ErrorCode>(err); -} - - -RdKafka::ErrorCode -RdKafka::KafkaConsumerImpl::unassign () { - return static_cast<RdKafka::ErrorCode>(rd_kafka_assign(rk_, NULL)); -} - - -RdKafka::ErrorCode -RdKafka::KafkaConsumerImpl::committed (std::vector<RdKafka::TopicPartition*> &partitions, int timeout_ms) { - rd_kafka_topic_partition_list_t *c_parts; - rd_kafka_resp_err_t err; - - c_parts = partitions_to_c_parts(partitions); - - err = rd_kafka_committed(rk_, c_parts, timeout_ms); - - if (!err) { - update_partitions_from_c_parts(partitions, c_parts); - } - - rd_kafka_topic_partition_list_destroy(c_parts); - - return static_cast<RdKafka::ErrorCode>(err); -} - - -RdKafka::ErrorCode -RdKafka::KafkaConsumerImpl::position (std::vector<RdKafka::TopicPartition*> &partitions) { - rd_kafka_topic_partition_list_t *c_parts; - rd_kafka_resp_err_t err; - - c_parts = partitions_to_c_parts(partitions); - - err = rd_kafka_position(rk_, c_parts); - - if (!err) { - update_partitions_from_c_parts(partitions, c_parts); - } - - rd_kafka_topic_partition_list_destroy(c_parts); - - return static_cast<RdKafka::ErrorCode>(err); -} - - -RdKafka::ErrorCode -RdKafka::KafkaConsumerImpl::seek (const RdKafka::TopicPartition &partition, - int timeout_ms) { - const RdKafka::TopicPartitionImpl *p = - dynamic_cast<const RdKafka::TopicPartitionImpl*>(&partition); - rd_kafka_topic_t *rkt; - - if (!(rkt = rd_kafka_topic_new(rk_, p->topic_.c_str(), NULL))) - return static_cast<RdKafka::ErrorCode>(rd_kafka_last_error()); - - /* FIXME: Use a C API that takes a topic_partition_list_t instead */ - RdKafka::ErrorCode err = - static_cast<RdKafka::ErrorCode> - (rd_kafka_seek(rkt, p->partition_, p->offset_, timeout_ms)); - - rd_kafka_topic_destroy(rkt); - - return err; -} - - - - - -RdKafka::ErrorCode -RdKafka::KafkaConsumerImpl::close () { - rd_kafka_resp_err_t err; - err = rd_kafka_consumer_close(rk_); - if (err) - return static_cast<RdKafka::ErrorCode>(err); - - while (rd_kafka_outq_len(rk_) > 0) - rd_kafka_poll(rk_, 10); - rd_kafka_destroy(rk_); - - return static_cast<RdKafka::ErrorCode>(err); -} - - - http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7528d23e/thirdparty/librdkafka-0.11.1/src-cpp/Makefile ---------------------------------------------------------------------- diff --git a/thirdparty/librdkafka-0.11.1/src-cpp/Makefile b/thirdparty/librdkafka-0.11.1/src-cpp/Makefile deleted file mode 100644 index 7b84b67..0000000 --- a/thirdparty/librdkafka-0.11.1/src-cpp/Makefile +++ /dev/null @@ -1,58 +0,0 @@ -PKGNAME= librdkafka -LIBNAME= librdkafka++ -LIBVER= 1 - -CXXSRCS= RdKafka.cpp ConfImpl.cpp HandleImpl.cpp \ - ConsumerImpl.cpp ProducerImpl.cpp KafkaConsumerImpl.cpp \ - TopicImpl.cpp TopicPartitionImpl.cpp MessageImpl.cpp \ - QueueImpl.cpp MetadataImpl.cpp - -HDRS= rdkafkacpp.h - -OBJS= $(CXXSRCS:%.cpp=%.o) - - - -all: lib check - - -include ../mklove/Makefile.base - -# No linker script/symbol hiding for C++ library -WITH_LDS=n - -# OSX and Cygwin requires linking required libraries -ifeq ($(_UNAME_S),Darwin) - FWD_LINKING_REQ=y -endif -ifeq ($(_UNAME_S),AIX) - FWD_LINKING_REQ=y -endif -ifeq ($(shell uname -o 2>/dev/null),Cygwin) - FWD_LINKING_REQ=y -endif - -# Ignore previously defined library dependencies for the C library, -# we'll get those dependencies through the C library linkage. -LIBS := -L../src -lrdkafka -lstdc++ - -CHECK_FILES+= $(LIBFILENAME) $(LIBNAME).a - - -file-check: lib -check: file-check - -install: lib-install - -clean: lib-clean - -ifeq ($(WITH_LDS),y) -# Enable linker script if supported by platform -LIB_LDFLAGS+= $(LDFLAG_LINKERSCRIPT)$(LIBNAME).lds -endif - -$(LIBNAME).lds: $(HDRS) - @(printf "$(MKL_YELLOW)Generating linker script $@ from $(HDRS)$(MKL_CLR_RESET)\n" ; \ - cat ../src/rdkafka.h | ../lds-gen.py > $@) - --include $(DEPS) http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7528d23e/thirdparty/librdkafka-0.11.1/src-cpp/MessageImpl.cpp ---------------------------------------------------------------------- diff --git a/thirdparty/librdkafka-0.11.1/src-cpp/MessageImpl.cpp b/thirdparty/librdkafka-0.11.1/src-cpp/MessageImpl.cpp deleted file mode 100644 index 9562402..0000000 --- a/thirdparty/librdkafka-0.11.1/src-cpp/MessageImpl.cpp +++ /dev/null @@ -1,38 +0,0 @@ -/* - * librdkafka - Apache Kafka C/C++ library - * - * Copyright (c) 2014 Magnus Edenhill - * All rights reserved. - * - * Redistribution and use in source and binary forms, with or without - * modification, are permitted provided that the following conditions are met: - * - * 1. Redistributions of source code must retain the above copyright notice, - * this list of conditions and the following disclaimer. - * 2. Redistributions in binary form must reproduce the above copyright notice, - * this list of conditions and the following disclaimer in the documentation - * and/or other materials provided with the distribution. - * - * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" - * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE - * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE - * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE - * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR - * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF - * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS - * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN - * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) - * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE - * POSSIBILITY OF SUCH DAMAGE. - */ - -#include <iostream> -#include <string> -#include <list> -#include <cerrno> - -#include "rdkafkacpp_int.h" - - -RdKafka::Message::~Message() {} - http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7528d23e/thirdparty/librdkafka-0.11.1/src-cpp/MetadataImpl.cpp ---------------------------------------------------------------------- diff --git a/thirdparty/librdkafka-0.11.1/src-cpp/MetadataImpl.cpp b/thirdparty/librdkafka-0.11.1/src-cpp/MetadataImpl.cpp deleted file mode 100644 index c2869f5..0000000 --- a/thirdparty/librdkafka-0.11.1/src-cpp/MetadataImpl.cpp +++ /dev/null @@ -1,151 +0,0 @@ -/* - * librdkafka - Apache Kafka C/C++ library - * - * Copyright (c) 2014 Magnus Edenhill - * All rights reserved. - * - * Redistribution and use in source and binary forms, with or without - * modification, are permitted provided that the following conditions are met: - * - * 1. Redistributions of source code must retain the above copyright notice, - * this list of conditions and the following disclaimer. - * 2. Redistributions in binary form must reproduce the above copyright notice, - * this list of conditions and the following disclaimer in the documentation - * and/or other materials provided with the distribution. - * - * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" - * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE - * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE - * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE - * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR - * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF - * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS - * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN - * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) - * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE - * POSSIBILITY OF SUCH DAMAGE. - */ - -#include "rdkafkacpp_int.h" - -using namespace RdKafka; - -BrokerMetadata::~BrokerMetadata() {}; -PartitionMetadata::~PartitionMetadata() {}; -TopicMetadata::~TopicMetadata() {}; -Metadata::~Metadata() {}; - - -/** - * Metadata: Broker information handler implementation - */ -class BrokerMetadataImpl : public BrokerMetadata { - public: - BrokerMetadataImpl(const rd_kafka_metadata_broker_t *broker_metadata) - :broker_metadata_(broker_metadata),host_(broker_metadata->host) {} - - int32_t id() const{return broker_metadata_->id;} - - const std::string host() const {return host_;} - int port() const {return broker_metadata_->port;} - - virtual ~BrokerMetadataImpl() {} - - private: - const rd_kafka_metadata_broker_t *broker_metadata_; - const std::string host_; -}; - -/** - * Metadata: Partition information handler - */ -class PartitionMetadataImpl : public PartitionMetadata { - public: - // @TODO too much memory copy? maybe we should create a new vector class that read directly from C arrays? - // @TODO use auto_ptr? - PartitionMetadataImpl(const rd_kafka_metadata_partition_t *partition_metadata) - :partition_metadata_(partition_metadata) { - replicas_.reserve(partition_metadata->replica_cnt); - for(int i=0;i<partition_metadata->replica_cnt;++i) - replicas_.push_back(partition_metadata->replicas[i]); - - isrs_.reserve(partition_metadata->isr_cnt); - for(int i=0;i<partition_metadata->isr_cnt;++i) - isrs_.push_back(partition_metadata->isrs[i]); - } - - int32_t id() const { - return partition_metadata_->id; - } - int32_t leader() const { - return partition_metadata_->leader; - } - ErrorCode err() const { - return static_cast<ErrorCode>(partition_metadata_->err); - } - - const std::vector<int32_t> *replicas() const {return &replicas_;} - const std::vector<int32_t> *isrs() const {return &isrs_;} - - ~PartitionMetadataImpl() {}; - - private: - const rd_kafka_metadata_partition_t *partition_metadata_; - std::vector<int32_t> replicas_,isrs_; -}; - -/** - * Metadata: Topic information handler - */ -class TopicMetadataImpl : public TopicMetadata{ - public: - TopicMetadataImpl(const rd_kafka_metadata_topic_t *topic_metadata) - :topic_metadata_(topic_metadata),topic_(topic_metadata->topic) { - partitions_.reserve(topic_metadata->partition_cnt); - for(int i=0;i<topic_metadata->partition_cnt;++i) - partitions_.push_back( - new PartitionMetadataImpl(&topic_metadata->partitions[i]) - ); - } - - ~TopicMetadataImpl(){ - for(size_t i=0;i<partitions_.size();++i) - delete partitions_[i]; - } - - const std::string topic() const {return topic_;} - const std::vector<const PartitionMetadata *> *partitions() const { - return &partitions_; - } - ErrorCode err() const {return static_cast<ErrorCode>(topic_metadata_->err);} - - private: - const rd_kafka_metadata_topic_t *topic_metadata_; - const std::string topic_; - std::vector<const PartitionMetadata *> partitions_; - -}; - -MetadataImpl::MetadataImpl(const rd_kafka_metadata_t *metadata) -:metadata_(metadata) -{ - brokers_.reserve(metadata->broker_cnt); - for(int i=0;i<metadata->broker_cnt;++i) - brokers_.push_back(new BrokerMetadataImpl(&metadata->brokers[i])); - - topics_.reserve(metadata->topic_cnt); - for(int i=0;i<metadata->topic_cnt;++i) - topics_.push_back(new TopicMetadataImpl(&metadata->topics[i])); - -} - -MetadataImpl::~MetadataImpl() { - for(size_t i=0;i<brokers_.size();++i) - delete brokers_[i]; - for(size_t i=0;i<topics_.size();++i) - delete topics_[i]; - - - if(metadata_) - rd_kafka_metadata_destroy(metadata_); -} http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7528d23e/thirdparty/librdkafka-0.11.1/src-cpp/ProducerImpl.cpp ---------------------------------------------------------------------- diff --git a/thirdparty/librdkafka-0.11.1/src-cpp/ProducerImpl.cpp b/thirdparty/librdkafka-0.11.1/src-cpp/ProducerImpl.cpp deleted file mode 100644 index e880573..0000000 --- a/thirdparty/librdkafka-0.11.1/src-cpp/ProducerImpl.cpp +++ /dev/null @@ -1,167 +0,0 @@ -/* - * librdkafka - Apache Kafka C/C++ library - * - * Copyright (c) 2014 Magnus Edenhill - * All rights reserved. - * - * Redistribution and use in source and binary forms, with or without - * modification, are permitted provided that the following conditions are met: - * - * 1. Redistributions of source code must retain the above copyright notice, - * this list of conditions and the following disclaimer. - * 2. Redistributions in binary form must reproduce the above copyright notice, - * this list of conditions and the following disclaimer in the documentation - * and/or other materials provided with the distribution. - * - * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" - * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE - * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE - * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE - * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR - * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF - * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS - * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN - * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) - * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE - * POSSIBILITY OF SUCH DAMAGE. - */ - -#include <iostream> -#include <string> -#include <list> -#include <cerrno> - -#include "rdkafkacpp_int.h" - - -RdKafka::Producer::~Producer () { - -} - -static void dr_msg_cb_trampoline (rd_kafka_t *rk, - const rd_kafka_message_t * - rkmessage, - void *opaque) { - RdKafka::HandleImpl *handle = static_cast<RdKafka::HandleImpl *>(opaque); - RdKafka::MessageImpl message(NULL, rkmessage); - handle->dr_cb_->dr_cb(message); -} - - - -RdKafka::Producer *RdKafka::Producer::create (RdKafka::Conf *conf, - std::string &errstr) { - char errbuf[512]; - RdKafka::ConfImpl *confimpl = dynamic_cast<RdKafka::ConfImpl *>(conf); - RdKafka::ProducerImpl *rkp = new RdKafka::ProducerImpl(); - rd_kafka_conf_t *rk_conf = NULL; - - if (confimpl) { - if (!confimpl->rk_conf_) { - errstr = "Requires RdKafka::Conf::CONF_GLOBAL object"; - delete rkp; - return NULL; - } - - rkp->set_common_config(confimpl); - - rk_conf = rd_kafka_conf_dup(confimpl->rk_conf_); - - if (confimpl->dr_cb_) { - rd_kafka_conf_set_dr_msg_cb(rk_conf, dr_msg_cb_trampoline); - rkp->dr_cb_ = confimpl->dr_cb_; - } - } - - - rd_kafka_t *rk; - if (!(rk = rd_kafka_new(RD_KAFKA_PRODUCER, rk_conf, - errbuf, sizeof(errbuf)))) { - errstr = errbuf; - delete rkp; - return NULL; - } - - rkp->rk_ = rk; - - return rkp; -} - - -RdKafka::ErrorCode RdKafka::ProducerImpl::produce (RdKafka::Topic *topic, - int32_t partition, - int msgflags, - void *payload, size_t len, - const std::string *key, - void *msg_opaque) { - RdKafka::TopicImpl *topicimpl = dynamic_cast<RdKafka::TopicImpl *>(topic); - - if (rd_kafka_produce(topicimpl->rkt_, partition, msgflags, - payload, len, - key ? key->c_str() : NULL, key ? key->size() : 0, - msg_opaque) == -1) - return static_cast<RdKafka::ErrorCode>(rd_kafka_last_error()); - - return RdKafka::ERR_NO_ERROR; -} - - -RdKafka::ErrorCode RdKafka::ProducerImpl::produce (RdKafka::Topic *topic, - int32_t partition, - int msgflags, - void *payload, size_t len, - const void *key, - size_t key_len, - void *msg_opaque) { - RdKafka::TopicImpl *topicimpl = dynamic_cast<RdKafka::TopicImpl *>(topic); - - if (rd_kafka_produce(topicimpl->rkt_, partition, msgflags, - payload, len, key, key_len, - msg_opaque) == -1) - return static_cast<RdKafka::ErrorCode>(rd_kafka_last_error()); - - return RdKafka::ERR_NO_ERROR; -} - - -RdKafka::ErrorCode -RdKafka::ProducerImpl::produce (RdKafka::Topic *topic, - int32_t partition, - const std::vector<char> *payload, - const std::vector<char> *key, - void *msg_opaque) { - RdKafka::TopicImpl *topicimpl = dynamic_cast<RdKafka::TopicImpl *>(topic); - - if (rd_kafka_produce(topicimpl->rkt_, partition, RD_KAFKA_MSG_F_COPY, - payload ? (void *)&(*payload)[0] : NULL, - payload ? payload->size() : 0, - key ? &(*key)[0] : NULL, key ? key->size() : 0, - msg_opaque) == -1) - return static_cast<RdKafka::ErrorCode>(rd_kafka_last_error()); - - return RdKafka::ERR_NO_ERROR; - -} - - -RdKafka::ErrorCode -RdKafka::ProducerImpl::produce (const std::string topic_name, - int32_t partition, int msgflags, - void *payload, size_t len, - const void *key, size_t key_len, - int64_t timestamp, - void *msg_opaque) { - return - static_cast<RdKafka::ErrorCode> - ( - rd_kafka_producev(rk_, - RD_KAFKA_V_TOPIC(topic_name.c_str()), - RD_KAFKA_V_PARTITION(partition), - RD_KAFKA_V_MSGFLAGS(msgflags), - RD_KAFKA_V_VALUE(payload, len), - RD_KAFKA_V_KEY(key, key_len), - RD_KAFKA_V_TIMESTAMP(timestamp), - RD_KAFKA_V_OPAQUE(msg_opaque), - RD_KAFKA_V_END) - ); -} http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7528d23e/thirdparty/librdkafka-0.11.1/src-cpp/QueueImpl.cpp ---------------------------------------------------------------------- diff --git a/thirdparty/librdkafka-0.11.1/src-cpp/QueueImpl.cpp b/thirdparty/librdkafka-0.11.1/src-cpp/QueueImpl.cpp deleted file mode 100644 index c64b1c1..0000000 --- a/thirdparty/librdkafka-0.11.1/src-cpp/QueueImpl.cpp +++ /dev/null @@ -1,66 +0,0 @@ -/* - * librdkafka - Apache Kafka C/C++ library - * - * Copyright (c) 2014 Magnus Edenhill - * All rights reserved. - * - * Redistribution and use in source and binary forms, with or without - * modification, are permitted provided that the following conditions are met: - * - * 1. Redistributions of source code must retain the above copyright notice, - * this list of conditions and the following disclaimer. - * 2. Redistributions in binary form must reproduce the above copyright notice, - * this list of conditions and the following disclaimer in the documentation - * and/or other materials provided with the distribution. - * - * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" - * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE - * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE - * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE - * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR - * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF - * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS - * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN - * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) - * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE - * POSSIBILITY OF SUCH DAMAGE. - */ - -#include <cerrno> - -#include "rdkafkacpp_int.h" - -RdKafka::Queue::~Queue () { - -} - -RdKafka::Queue *RdKafka::Queue::create (Handle *base) { - RdKafka::QueueImpl *queueimpl = new RdKafka::QueueImpl; - queueimpl->queue_ = rd_kafka_queue_new(dynamic_cast<HandleImpl*>(base)->rk_); - return queueimpl; -} - -RdKafka::ErrorCode -RdKafka::QueueImpl::forward (Queue *queue) { - if (!queue) { - rd_kafka_queue_forward(queue_, NULL); - } else { - QueueImpl *queueimpl = dynamic_cast<QueueImpl *>(queue); - rd_kafka_queue_forward(queue_, queueimpl->queue_); - } - return RdKafka::ERR_NO_ERROR; -} - -RdKafka::Message *RdKafka::QueueImpl::consume (int timeout_ms) { - rd_kafka_message_t *rkmessage; - rkmessage = rd_kafka_consume_queue(queue_, timeout_ms); - - if (!rkmessage) - return new RdKafka::MessageImpl(NULL, RdKafka::ERR__TIMED_OUT); - - return new RdKafka::MessageImpl(rkmessage); -} - -int RdKafka::QueueImpl::poll (int timeout_ms) { - return rd_kafka_queue_poll_callback(queue_, timeout_ms); -} http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7528d23e/thirdparty/librdkafka-0.11.1/src-cpp/README.md ---------------------------------------------------------------------- diff --git a/thirdparty/librdkafka-0.11.1/src-cpp/README.md b/thirdparty/librdkafka-0.11.1/src-cpp/README.md deleted file mode 100644 index a484589..0000000 --- a/thirdparty/librdkafka-0.11.1/src-cpp/README.md +++ /dev/null @@ -1,16 +0,0 @@ -librdkafka C++ interface -======================== - -**See rdkafkacpp.h for the public C++ API** - - - -Maintainer notes for the C++ interface: - - * The public C++ interface (rdkafkacpp.h) does not include the - public C interface (rdkafka.h) in any way, this means that all - constants, flags, etc, must be kept in sync manually between the two - header files. - A regression test should be implemented that checks this is true. - - * The public C++ interface is provided using pure virtual abstract classes. http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7528d23e/thirdparty/librdkafka-0.11.1/src-cpp/RdKafka.cpp ---------------------------------------------------------------------- diff --git a/thirdparty/librdkafka-0.11.1/src-cpp/RdKafka.cpp b/thirdparty/librdkafka-0.11.1/src-cpp/RdKafka.cpp deleted file mode 100644 index 7b67a7b..0000000 --- a/thirdparty/librdkafka-0.11.1/src-cpp/RdKafka.cpp +++ /dev/null @@ -1,52 +0,0 @@ -/* - * librdkafka - Apache Kafka C/C++ library - * - * Copyright (c) 2014 Magnus Edenhill - * All rights reserved. - * - * Redistribution and use in source and binary forms, with or without - * modification, are permitted provided that the following conditions are met: - * - * 1. Redistributions of source code must retain the above copyright notice, - * this list of conditions and the following disclaimer. - * 2. Redistributions in binary form must reproduce the above copyright notice, - * this list of conditions and the following disclaimer in the documentation - * and/or other materials provided with the distribution. - * - * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" - * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE - * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE - * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE - * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR - * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF - * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS - * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN - * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) - * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE - * POSSIBILITY OF SUCH DAMAGE. - */ - -#include <string> - -#include "rdkafkacpp_int.h" - -int RdKafka::version () { - return rd_kafka_version(); -} - -std::string RdKafka::version_str () { - return std::string(rd_kafka_version_str()); -} - -std::string RdKafka::get_debug_contexts() { - return std::string(RD_KAFKA_DEBUG_CONTEXTS); -} - -std::string RdKafka::err2str (RdKafka::ErrorCode err) { - return std::string(rd_kafka_err2str(static_cast<rd_kafka_resp_err_t>(err))); -} - -int RdKafka::wait_destroyed (int timeout_ms) { - return rd_kafka_wait_destroyed(timeout_ms); -} - http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7528d23e/thirdparty/librdkafka-0.11.1/src-cpp/TopicImpl.cpp ---------------------------------------------------------------------- diff --git a/thirdparty/librdkafka-0.11.1/src-cpp/TopicImpl.cpp b/thirdparty/librdkafka-0.11.1/src-cpp/TopicImpl.cpp deleted file mode 100644 index f330513..0000000 --- a/thirdparty/librdkafka-0.11.1/src-cpp/TopicImpl.cpp +++ /dev/null @@ -1,124 +0,0 @@ -/* - * librdkafka - Apache Kafka C/C++ library - * - * Copyright (c) 2014 Magnus Edenhill - * All rights reserved. - * - * Redistribution and use in source and binary forms, with or without - * modification, are permitted provided that the following conditions are met: - * - * 1. Redistributions of source code must retain the above copyright notice, - * this list of conditions and the following disclaimer. - * 2. Redistributions in binary form must reproduce the above copyright notice, - * this list of conditions and the following disclaimer in the documentation - * and/or other materials provided with the distribution. - * - * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" - * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE - * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE - * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE - * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR - * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF - * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS - * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN - * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) - * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE - * POSSIBILITY OF SUCH DAMAGE. - */ - -#include <iostream> -#include <string> -#include <list> -#include <cerrno> - -#include "rdkafkacpp_int.h" - -const int32_t RdKafka::Topic::PARTITION_UA = RD_KAFKA_PARTITION_UA; - -const int64_t RdKafka::Topic::OFFSET_BEGINNING = RD_KAFKA_OFFSET_BEGINNING; - -const int64_t RdKafka::Topic::OFFSET_END = RD_KAFKA_OFFSET_END; - -const int64_t RdKafka::Topic::OFFSET_STORED = RD_KAFKA_OFFSET_STORED; - -const int64_t RdKafka::Topic::OFFSET_INVALID = RD_KAFKA_OFFSET_INVALID; - -RdKafka::Topic::~Topic () { - -} - -static int32_t partitioner_cb_trampoline (const rd_kafka_topic_t *rkt, - const void *keydata, - size_t keylen, - int32_t partition_cnt, - void *rkt_opaque, - void *msg_opaque) { - RdKafka::TopicImpl *topicimpl = static_cast<RdKafka::TopicImpl *>(rkt_opaque); - std::string key(static_cast<const char *>(keydata), keylen); - return topicimpl->partitioner_cb_->partitioner_cb(topicimpl, &key, - partition_cnt, msg_opaque); -} - -static int32_t partitioner_kp_cb_trampoline (const rd_kafka_topic_t *rkt, - const void *keydata, - size_t keylen, - int32_t partition_cnt, - void *rkt_opaque, - void *msg_opaque) { - RdKafka::TopicImpl *topicimpl = static_cast<RdKafka::TopicImpl *>(rkt_opaque); - return topicimpl->partitioner_kp_cb_->partitioner_cb(topicimpl, - keydata, keylen, - partition_cnt, - msg_opaque); -} - - - -RdKafka::Topic *RdKafka::Topic::create (Handle *base, - const std::string &topic_str, - Conf *conf, - std::string &errstr) { - RdKafka::ConfImpl *confimpl = static_cast<RdKafka::ConfImpl *>(conf); - rd_kafka_topic_t *rkt; - rd_kafka_topic_conf_t *rkt_conf; - - RdKafka::TopicImpl *topic = new RdKafka::TopicImpl(); - - if (!confimpl) - rkt_conf = rd_kafka_topic_conf_new(); - else /* Make a copy of conf struct to allow Conf reuse. */ - rkt_conf = rd_kafka_topic_conf_dup(confimpl->rkt_conf_); - - /* Set topic opaque to the topic so that we can reach our topic object - * from whatever callbacks get registered. - * The application itself will not need these opaques since their - * callbacks are class based. */ - rd_kafka_topic_conf_set_opaque(rkt_conf, static_cast<void *>(topic)); - - if (confimpl) { - if (confimpl->partitioner_cb_) { - rd_kafka_topic_conf_set_partitioner_cb(rkt_conf, - partitioner_cb_trampoline); - topic->partitioner_cb_ = confimpl->partitioner_cb_; - } else if (confimpl->partitioner_kp_cb_) { - rd_kafka_topic_conf_set_partitioner_cb(rkt_conf, - partitioner_kp_cb_trampoline); - topic->partitioner_kp_cb_ = confimpl->partitioner_kp_cb_; - } - } - - - if (!(rkt = rd_kafka_topic_new(dynamic_cast<HandleImpl*>(base)->rk_, - topic_str.c_str(), rkt_conf))) { - errstr = rd_kafka_err2str(rd_kafka_last_error()); - delete topic; - rd_kafka_topic_conf_destroy(rkt_conf); - return NULL; - } - - topic->rkt_ = rkt; - - return topic; - -} - http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/7528d23e/thirdparty/librdkafka-0.11.1/src-cpp/TopicPartitionImpl.cpp ---------------------------------------------------------------------- diff --git a/thirdparty/librdkafka-0.11.1/src-cpp/TopicPartitionImpl.cpp b/thirdparty/librdkafka-0.11.1/src-cpp/TopicPartitionImpl.cpp deleted file mode 100644 index 71a688c..0000000 --- a/thirdparty/librdkafka-0.11.1/src-cpp/TopicPartitionImpl.cpp +++ /dev/null @@ -1,55 +0,0 @@ -/* - * librdkafka - Apache Kafka C/C++ library - * - * Copyright (c) 2015 Magnus Edenhill - * All rights reserved. - * - * Redistribution and use in source and binary forms, with or without - * modification, are permitted provided that the following conditions are met: - * - * 1. Redistributions of source code must retain the above copyright notice, - * this list of conditions and the following disclaimer. - * 2. Redistributions in binary form must reproduce the above copyright notice, - * this list of conditions and the following disclaimer in the documentation - * and/or other materials provided with the distribution. - * - * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" - * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE - * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE - * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE - * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR - * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF - * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS - * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN - * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) - * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE - * POSSIBILITY OF SUCH DAMAGE. - */ - -#include <iostream> -#include <string> -#include <vector> - -#include "rdkafkacpp_int.h" - -RdKafka::TopicPartition::~TopicPartition () { -} - -RdKafka::TopicPartition * -RdKafka::TopicPartition::create (const std::string &topic, int partition) { - return new TopicPartitionImpl(topic, partition); -} - -RdKafka::TopicPartition * -RdKafka::TopicPartition::create (const std::string &topic, int partition, - int64_t offset) { - return new TopicPartitionImpl(topic, partition, offset); -} - -void -RdKafka::TopicPartition::destroy (std::vector<TopicPartition*> &partitions) { - for (std::vector<TopicPartition*>::iterator it = partitions.begin() ; - it != partitions.end(); ++it) - delete(*it); - partitions.clear(); -}
