[GitHub] nifi-minifi-cpp pull request #437: MINIFICPP-558: initial provisioning for C...
GitHub user phrocker opened a pull request: https://github.com/apache/nifi-minifi-cpp/pull/437 MINIFICPP-558: initial provisioning for CoAP Thank you for submitting a contribution to Apache NiFi - MiNiFi C++. In order to streamline the review of the contribution we ask you to ensure the following steps have been taken: ### For all changes: - [ ] Is there a JIRA ticket associated with this PR? Is it referenced in the commit message? - [ ] Does your PR title start with MINIFICPP- where is the JIRA number you are trying to resolve? Pay particular attention to the hyphen "-" character. - [ ] Has your PR been rebased against the latest commit within the target branch (typically master)? - [ ] Is your initial contribution a single, squashed commit? ### For code changes: - [ ] If adding new dependencies to the code, are these dependencies licensed in a way that is compatible for inclusion under [ASF 2.0](http://www.apache.org/legal/resolved.html#category-a)? - [ ] If applicable, have you updated the LICENSE file? - [ ] If applicable, have you updated the NOTICE file? ### For documentation related changes: - [ ] Have you ensured that format looks appropriate for the output in which it is rendered? ### Note: Please ensure that once the PR is submitted, you check travis-ci for build issues and submit an update to your PR as soon as possible. You can merge this pull request into a Git repository by running: $ git pull https://github.com/phrocker/nifi-minifi-cpp MINIFICPP-558 Alternatively you can review and apply these changes as the patch at: https://github.com/apache/nifi-minifi-cpp/pull/437.patch To close this pull request, make a commit to your master/trunk branch with (at least) the following in the commit message: This closes #437 commit bee4aad6c91b1926e90ac9ce5646e04e865410cc Author: Marc Parisi Date: 2018-10-23T15:51:19Z MINIFICPP-558: initial provisioning for CoAP ---
[GitHub] nifi-minifi-cpp pull request #437: MINIFICPP-558: initial provisioning for C...
Github user phrocker commented on a diff in the pull request: https://github.com/apache/nifi-minifi-cpp/pull/437#discussion_r233474220 --- Diff: extensions/coap/CMakeLists.txt --- @@ -0,0 +1,92 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +# + + +include(${CMAKE_SOURCE_DIR}/extensions/ExtensionHeader.txt) +include_directories(protocols nanofi controllerservice) +include_directories(../http-curl/) + +file(GLOB CSOURCES "nanofi/*.c") +file(GLOB SOURCES "*.cpp" "protocols/*.cpp" "processors/*.cpp" "controllerservice/*.cpp" ) + +add_library(nanofi-coap-c STATIC ${CSOURCES}) +add_library(minifi-coap STATIC ${SOURCES}) +set_property(TARGET minifi-coap PROPERTY POSITION_INDEPENDENT_CODE ON) + +if(CMAKE_THREAD_LIBS_INIT) + target_link_libraries(minifi-coap "${CMAKE_THREAD_LIBS_INIT}") +endif() + + set(BASE_DIR "${CMAKE_CURRENT_BINARY_DIR}/extensions/coap") + if (APPLE) + set(BYPRODUCT "${BASE_DIR}/extensions/coap/thirdparty/libcoap-src/.libs/libcoap-2-gnutls.a") --- End diff -- this may need a little modification as artifacts can vary depending on what is used for building TLS ---
[GitHub] nifi-minifi-cpp pull request #437: MINIFICPP-558: initial provisioning for C...
Github user arpadboda commented on a diff in the pull request: https://github.com/apache/nifi-minifi-cpp/pull/437#discussion_r233497401 --- Diff: docker/test/integration/minifi/test/__init__.py --- @@ -42,6 +42,8 @@ def __init__(self, output_validator): self.segfault = False +self.segfault = False --- End diff -- This is already there two lines above. ---
[GitHub] nifi-minifi-cpp pull request #437: MINIFICPP-558: initial provisioning for C...
Github user arpadboda commented on a diff in the pull request: https://github.com/apache/nifi-minifi-cpp/pull/437#discussion_r233499744 --- Diff: extensions/coap/controllerservice/CoapConnector.h --- @@ -0,0 +1,207 @@ +/** + * + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +#ifndef LIBMINIFI_INCLUDE_CONTROLLERS_COAPCONNECTOR_H_ +#define LIBMINIFI_INCLUDE_CONTROLLERS_COAPCONNECTOR_H_ + + +#include "core/logging/LoggerConfiguration.h" +#include "coap_functions.h" +#include "core/controller/ControllerService.h" +#include +#include + +namespace org { +namespace apache { +namespace nifi { +namespace minifi { +namespace controllers { + +/** + * Purpose and Justification: Controller services function as a layerable way to provide + * services to internal services. While a controller service is generally configured from the flow, + * we want to follow the open closed principle and provide CoAP services to other components. + * + * + */ +class CoapConnectorService : public core::controller::ControllerService { + public: + + /** + * CoapMessage is in internal message format that is sent to and from consumers of this controller service. + */ + class CoAPMessage { + public: + +explicit CoAPMessage(unsigned int code = 0) +: code_(code) { +} + +explicit CoAPMessage(unsigned int code, unsigned char *data, size_t dataLen) +: code_(code) { + if (data && dataLen > 0) +std::copy(data, data + dataLen, std::back_inserter(data_)); +} + +CoAPMessage(const CoAPMessage &other) = delete; + +CoAPMessage(CoAPMessage &&other) = default; + +~CoAPMessage() { +} + +size_t getSize() const { + return data_.size(); +} +unsigned char const *getData() const { + return data_.data(); +} + +bool isRegistrationRequest() { + if (data_.size() != 8) { +return false; + } + return code_ == COAP_RESPONSE_400 && std::string((char*) data_.data(), data_.size()) == "register"; +} +CoAPMessage &operator=(const CoAPMessage &other) = delete; +CoAPMessage &operator=(CoAPMessage &&other) = default; + private: +unsigned int code_; +std::vector data_; + }; + + /** + * Constructors for the controller service. + */ + explicit CoapConnectorService(const std::string &name, const std::string &id) + : ControllerService(name, id), +port_(0), +initialized_(false), +logger_(logging::LoggerFactory::getLogger()) { +initialize(); + } + + explicit CoapConnectorService(const std::string &name, utils::Identifier uuid = utils::Identifier()) + : ControllerService(name, uuid), +port_(0), +initialized_(false), +logger_(logging::LoggerFactory::getLogger()) { +initialize(); + } + + explicit CoapConnectorService(const std::string &name, const std::shared_ptr &configuration) + : ControllerService(name), +port_(0), +initialized_(false), +logger_(logging::LoggerFactory::getLogger()) { +setConfiguration(configuration); +initialize(); + } + + /** + * Parameters needed. + */ + static core::Property RemoteServer; + static core::Property Port; + static core::Property MaxQueueSize; + + virtual void initialize(); + + void yield() { + + } + + bool isRunning() { +return getState() == core::controller::ControllerServiceState::ENABLED; + } + + bool isWorkAvailable() { +return false; + } + + virtual void onEnable(); + + /** + * Sends the payload to the endpoint, returning the response as we await. Will retry transmission + * @param type type of payload to endpoint interaction ( GET, PO
[GitHub] nifi-minifi-cpp pull request #437: MINIFICPP-558: initial provisioning for C...
Github user arpadboda commented on a diff in the pull request: https://github.com/apache/nifi-minifi-cpp/pull/437#discussion_r233500462 --- Diff: extensions/coap/nanofi/coap_functions.c --- @@ -0,0 +1,175 @@ +/** + * + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +#include "coap_functions.h" + + +#ifdef __cplusplus +extern "C" { +#endif + +/** + * Initialize the API access. Not thread safe. + */ +void init_coap_api(void *rcvr, callback_pointers *ptrs) { + global_ptrs.data_received = ptrs->data_received; + global_ptrs.received_error = ptrs->received_error; + receiver = rcvr; +} + + +int create_session(coap_context_t **ctx, coap_session_t **session, const char *node, const char *port, coap_address_t *dst_addr) { + int s; + struct addrinfo hints; + coap_proto_t proto = COAP_PROTO_UDP; + struct addrinfo *result, *rp; + + memset(&hints, 0, sizeof(struct addrinfo)); + hints.ai_family = AF_UNSPEC; // ipv4 or ipv6 + hints.ai_socktype = COAP_PROTO_RELIABLE(proto) ? SOCK_STREAM : SOCK_DGRAM; + hints.ai_flags = AI_PASSIVE | AI_NUMERICHOST | AI_NUMERICSERV | AI_ALL; + + s = getaddrinfo(node, port, &hints, &result); + if (s != 0) { +fprintf(stderr, "getaddrinfo: %s\n", gai_strerror(s)); +return -1; + } + + for (rp = result; rp != NULL; rp = rp->ai_next) { +coap_address_t addr; + +if (rp->ai_addrlen <= sizeof(addr.addr)) { + coap_address_init(&addr); + addr.size = rp->ai_addrlen; + memcpy(&addr.addr, rp->ai_addr, rp->ai_addrlen); + + *ctx = coap_new_context(0x00); + + *session = coap_new_client_session(*ctx, &addr, dst_addr, proto); + if (*ctx && *session) { +freeaddrinfo(result); +return 0; + } +} + } + + fprintf(stderr, "no context available for interface '%s'\n", node); + + freeaddrinfo(result); + return -1; +} + +struct coap_pdu_t *create_request(struct coap_context_t *ctx,struct coap_session_t *session,coap_optlist_t **optlist, unsigned char code, coap_str_const_t *ptr) { + coap_pdu_t *pdu; + + if (!(pdu = coap_new_pdu(session))) +return NULL; + + pdu->type = COAP_MESSAGE_CON; + pdu->tid = coap_new_message_id(session); + pdu->code = code; + + if (optlist){ +coap_add_optlist_pdu(pdu, optlist); + } + + int flags = 0; + coap_add_data(pdu, ptr->length, ptr->s); + return pdu; +} + +int coap_event(struct coap_context_t *ctx, coap_event_t event, struct coap_session_t *session){ +if (event == COAP_EVENT_SESSION_FAILED && global_ptrs.received_error){ +global_ptrs.received_error(receiver, ctx, -1); +} +return 0; +} + +void no_acknowledgement(struct coap_context_t *ctx, coap_session_t *session, coap_pdu_t *sent, coap_nack_reason_t reason, const coap_tid_t id){ + if (global_ptrs.received_error){ + global_ptrs.received_error(receiver, ctx, -1); --- End diff -- Indentation ---
[GitHub] nifi-minifi-cpp pull request #437: MINIFICPP-558: initial provisioning for C...
Github user arpadboda commented on a diff in the pull request: https://github.com/apache/nifi-minifi-cpp/pull/437#discussion_r233500619 --- Diff: extensions/coap/nanofi/coap_functions.c --- @@ -0,0 +1,175 @@ +/** + * + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +#include "coap_functions.h" + + +#ifdef __cplusplus +extern "C" { +#endif + +/** + * Initialize the API access. Not thread safe. + */ +void init_coap_api(void *rcvr, callback_pointers *ptrs) { + global_ptrs.data_received = ptrs->data_received; + global_ptrs.received_error = ptrs->received_error; + receiver = rcvr; +} + + +int create_session(coap_context_t **ctx, coap_session_t **session, const char *node, const char *port, coap_address_t *dst_addr) { + int s; + struct addrinfo hints; + coap_proto_t proto = COAP_PROTO_UDP; + struct addrinfo *result, *rp; + + memset(&hints, 0, sizeof(struct addrinfo)); + hints.ai_family = AF_UNSPEC; // ipv4 or ipv6 + hints.ai_socktype = COAP_PROTO_RELIABLE(proto) ? SOCK_STREAM : SOCK_DGRAM; + hints.ai_flags = AI_PASSIVE | AI_NUMERICHOST | AI_NUMERICSERV | AI_ALL; + + s = getaddrinfo(node, port, &hints, &result); + if (s != 0) { +fprintf(stderr, "getaddrinfo: %s\n", gai_strerror(s)); +return -1; + } + + for (rp = result; rp != NULL; rp = rp->ai_next) { +coap_address_t addr; + +if (rp->ai_addrlen <= sizeof(addr.addr)) { + coap_address_init(&addr); + addr.size = rp->ai_addrlen; + memcpy(&addr.addr, rp->ai_addr, rp->ai_addrlen); + + *ctx = coap_new_context(0x00); + + *session = coap_new_client_session(*ctx, &addr, dst_addr, proto); + if (*ctx && *session) { +freeaddrinfo(result); +return 0; + } +} + } + + fprintf(stderr, "no context available for interface '%s'\n", node); + + freeaddrinfo(result); + return -1; +} + +struct coap_pdu_t *create_request(struct coap_context_t *ctx,struct coap_session_t *session,coap_optlist_t **optlist, unsigned char code, coap_str_const_t *ptr) { + coap_pdu_t *pdu; + + if (!(pdu = coap_new_pdu(session))) +return NULL; + + pdu->type = COAP_MESSAGE_CON; + pdu->tid = coap_new_message_id(session); + pdu->code = code; + + if (optlist){ +coap_add_optlist_pdu(pdu, optlist); + } + + int flags = 0; + coap_add_data(pdu, ptr->length, ptr->s); + return pdu; +} + +int coap_event(struct coap_context_t *ctx, coap_event_t event, struct coap_session_t *session){ +if (event == COAP_EVENT_SESSION_FAILED && global_ptrs.received_error){ --- End diff -- Please indent with 2 spaces ---
[GitHub] nifi-minifi-cpp pull request #437: MINIFICPP-558: initial provisioning for C...
Github user arpadboda commented on a diff in the pull request: https://github.com/apache/nifi-minifi-cpp/pull/437#discussion_r233500950 --- Diff: extensions/coap/nanofi/coap_functions.c --- @@ -0,0 +1,175 @@ +/** + * + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +#include "coap_functions.h" + + +#ifdef __cplusplus +extern "C" { +#endif + +/** + * Initialize the API access. Not thread safe. + */ +void init_coap_api(void *rcvr, callback_pointers *ptrs) { + global_ptrs.data_received = ptrs->data_received; + global_ptrs.received_error = ptrs->received_error; + receiver = rcvr; +} + + +int create_session(coap_context_t **ctx, coap_session_t **session, const char *node, const char *port, coap_address_t *dst_addr) { + int s; + struct addrinfo hints; + coap_proto_t proto = COAP_PROTO_UDP; + struct addrinfo *result, *rp; + + memset(&hints, 0, sizeof(struct addrinfo)); + hints.ai_family = AF_UNSPEC; // ipv4 or ipv6 + hints.ai_socktype = COAP_PROTO_RELIABLE(proto) ? SOCK_STREAM : SOCK_DGRAM; + hints.ai_flags = AI_PASSIVE | AI_NUMERICHOST | AI_NUMERICSERV | AI_ALL; + + s = getaddrinfo(node, port, &hints, &result); + if (s != 0) { +fprintf(stderr, "getaddrinfo: %s\n", gai_strerror(s)); +return -1; + } + + for (rp = result; rp != NULL; rp = rp->ai_next) { +coap_address_t addr; + +if (rp->ai_addrlen <= sizeof(addr.addr)) { + coap_address_init(&addr); + addr.size = rp->ai_addrlen; + memcpy(&addr.addr, rp->ai_addr, rp->ai_addrlen); + + *ctx = coap_new_context(0x00); + + *session = coap_new_client_session(*ctx, &addr, dst_addr, proto); + if (*ctx && *session) { +freeaddrinfo(result); +return 0; + } +} + } + + fprintf(stderr, "no context available for interface '%s'\n", node); + + freeaddrinfo(result); + return -1; +} + +struct coap_pdu_t *create_request(struct coap_context_t *ctx,struct coap_session_t *session,coap_optlist_t **optlist, unsigned char code, coap_str_const_t *ptr) { + coap_pdu_t *pdu; + + if (!(pdu = coap_new_pdu(session))) +return NULL; + + pdu->type = COAP_MESSAGE_CON; + pdu->tid = coap_new_message_id(session); + pdu->code = code; + + if (optlist){ +coap_add_optlist_pdu(pdu, optlist); + } + + int flags = 0; + coap_add_data(pdu, ptr->length, ptr->s); + return pdu; +} + +int coap_event(struct coap_context_t *ctx, coap_event_t event, struct coap_session_t *session){ +if (event == COAP_EVENT_SESSION_FAILED && global_ptrs.received_error){ +global_ptrs.received_error(receiver, ctx, -1); +} +return 0; +} + +void no_acknowledgement(struct coap_context_t *ctx, coap_session_t *session, coap_pdu_t *sent, coap_nack_reason_t reason, const coap_tid_t id){ + if (global_ptrs.received_error){ + global_ptrs.received_error(receiver, ctx, -1); + } +} + +void response_handler(struct coap_context_t *ctx, struct coap_session_t *session, coap_pdu_t *sent, coap_pdu_t *received, const coap_tid_t id) { + unsigned char* data; + size_t data_len; + coap_opt_iterator_t opt_iter; + coap_opt_t * block_opt = coap_check_option(received, COAP_OPTION_BLOCK1, &opt_iter); + if (block_opt) { --- End diff -- Why not ``` if(!block_opt) ``` ? ---
[GitHub] nifi-minifi-cpp pull request #437: MINIFICPP-558: initial provisioning for C...
Github user arpadboda commented on a diff in the pull request: https://github.com/apache/nifi-minifi-cpp/pull/437#discussion_r233501997 --- Diff: extensions/coap/nanofi/coap_functions.c --- @@ -0,0 +1,175 @@ +/** + * + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +#include "coap_functions.h" + + +#ifdef __cplusplus +extern "C" { +#endif + +/** + * Initialize the API access. Not thread safe. + */ +void init_coap_api(void *rcvr, callback_pointers *ptrs) { + global_ptrs.data_received = ptrs->data_received; + global_ptrs.received_error = ptrs->received_error; + receiver = rcvr; +} + + +int create_session(coap_context_t **ctx, coap_session_t **session, const char *node, const char *port, coap_address_t *dst_addr) { + int s; + struct addrinfo hints; + coap_proto_t proto = COAP_PROTO_UDP; + struct addrinfo *result, *rp; + + memset(&hints, 0, sizeof(struct addrinfo)); + hints.ai_family = AF_UNSPEC; // ipv4 or ipv6 + hints.ai_socktype = COAP_PROTO_RELIABLE(proto) ? SOCK_STREAM : SOCK_DGRAM; + hints.ai_flags = AI_PASSIVE | AI_NUMERICHOST | AI_NUMERICSERV | AI_ALL; + + s = getaddrinfo(node, port, &hints, &result); + if (s != 0) { +fprintf(stderr, "getaddrinfo: %s\n", gai_strerror(s)); +return -1; + } + + for (rp = result; rp != NULL; rp = rp->ai_next) { +coap_address_t addr; + +if (rp->ai_addrlen <= sizeof(addr.addr)) { + coap_address_init(&addr); + addr.size = rp->ai_addrlen; + memcpy(&addr.addr, rp->ai_addr, rp->ai_addrlen); + + *ctx = coap_new_context(0x00); + + *session = coap_new_client_session(*ctx, &addr, dst_addr, proto); + if (*ctx && *session) { +freeaddrinfo(result); +return 0; + } +} + } + + fprintf(stderr, "no context available for interface '%s'\n", node); + + freeaddrinfo(result); + return -1; +} + +struct coap_pdu_t *create_request(struct coap_context_t *ctx,struct coap_session_t *session,coap_optlist_t **optlist, unsigned char code, coap_str_const_t *ptr) { + coap_pdu_t *pdu; + + if (!(pdu = coap_new_pdu(session))) +return NULL; + + pdu->type = COAP_MESSAGE_CON; + pdu->tid = coap_new_message_id(session); + pdu->code = code; + + if (optlist){ +coap_add_optlist_pdu(pdu, optlist); + } + + int flags = 0; + coap_add_data(pdu, ptr->length, ptr->s); + return pdu; +} + +int coap_event(struct coap_context_t *ctx, coap_event_t event, struct coap_session_t *session){ +if (event == COAP_EVENT_SESSION_FAILED && global_ptrs.received_error){ +global_ptrs.received_error(receiver, ctx, -1); +} +return 0; +} + +void no_acknowledgement(struct coap_context_t *ctx, coap_session_t *session, coap_pdu_t *sent, coap_nack_reason_t reason, const coap_tid_t id){ + if (global_ptrs.received_error){ + global_ptrs.received_error(receiver, ctx, -1); + } +} + +void response_handler(struct coap_context_t *ctx, struct coap_session_t *session, coap_pdu_t *sent, coap_pdu_t *received, const coap_tid_t id) { + unsigned char* data; + size_t data_len; + coap_opt_iterator_t opt_iter; + coap_opt_t * block_opt = coap_check_option(received, COAP_OPTION_BLOCK1, &opt_iter); + if (block_opt) { + } else { +if (!global_ptrs.data_received){ + return; +} + +if (COAP_RESPONSE_CLASS(received->code) == 2 || received->code == COAP_RESPONSE_400) { + if (coap_get_data(received, &data_len, &data)) { +if (global_ptrs.data_received) +global_ptrs.data_received(receiver, ctx, received->code, data, &data_len); + } +} +else{ + if (global_ptrs.received_error) +global_ptrs.received_error(receiver, ctx, received->code); +} + } + +} + +int resolve_address(
[GitHub] nifi-minifi-cpp pull request #437: MINIFICPP-558: initial provisioning for C...
Github user arpadboda commented on a diff in the pull request: https://github.com/apache/nifi-minifi-cpp/pull/437#discussion_r233504886 --- Diff: extensions/coap/protocols/CoapC2Protocol.cpp --- @@ -0,0 +1,353 @@ +/** + * + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +#include "CoapC2Protocol.h" +#include "c2/PayloadSerializer.h" + +namespace org { +namespace apache { +namespace nifi { +namespace minifi { +namespace c2 { + +#include "coap_functions.h" +#include "io/BaseStream.h" + +CoapProtocol::CoapProtocol(std::string name, utils::Identifier uuid) +: RESTSender(name, uuid), + require_registration_(false), + logger_(logging::LoggerFactory::getLogger()) { +} + +CoapProtocol::~CoapProtocol() { +} + +void CoapProtocol::initialize(const std::shared_ptr &controller, const std::shared_ptr &configure) { + RESTSender::initialize(controller, configure); + if (configure->get("nifi.c2.coap.connector.service", controller_service_name_)) { +auto service = controller->getControllerService(controller_service_name_); +coap_service_ = std::static_pointer_cast(service); + } else { +logger_->log_info("No CoAP connector configured, so using default service"); +coap_service_ = std::make_shared("cs", configure); +coap_service_->onEnable(); + } +} + +C2Payload CoapProtocol::consumePayload(const std::string &url, const C2Payload &payload, Direction direction, bool async) { + return RESTSender::consumePayload(url, payload, direction, false); +} + +int CoapProtocol::writeAcknowledgement(io::BaseStream *stream, const C2Payload &payload) { + auto ident = payload.getIdentifier(); + auto state = payload.getStatus().getState(); + stream->writeUTF(ident); + uint8_t payloadState = 0; + switch (state) { +case state::UpdateState::NESTED: +case state::UpdateState::INITIATE: +case state::UpdateState::FULLY_APPLIED: +case state::UpdateState::READ_COMPLETE: + payloadState = 0; + break; +case state::UpdateState::NOT_APPLIED: +case state::UpdateState::PARTIALLY_APPLIED: + payloadState = 1; + break; +case state::UpdateState::READ_ERROR: + payloadState = 2; + break; +case state::UpdateState::SET_ERROR: + payloadState = 3; + break; + } + stream->write(&payloadState, 1); + return 0; +} + +int CoapProtocol::writeHeartbeat(io::BaseStream *stream, const C2Payload &payload) { + bool byte; + uint16_t size = 0; + + std::string deviceIdent; + // device identifier + auto deviceInfo = getPayload("deviceInfo", payload); + if (deviceInfo) { +for (const auto &component : deviceInfo->getContent()) { + if (!getString(&component, "identifier", &deviceIdent)) { +break; + } +} + } + stream->writeUTF(deviceIdent, false); + std::string agentIdent; + // agent identifier + auto agentInfo = getPayload("agentInfo", payload); + if (agentInfo) { +for (const auto &component : agentInfo->getContent()) { + if (!getString(&component, "identifier", &agentIdent)) { +break; + } +} + } + if (agentIdent.empty()) { +return -1; + } + stream->writeUTF(agentIdent, false); + + auto flowInfo = getPayload("flowInfo", payload); + + if (flowInfo != nullptr) { + +auto components = getPayload("components", flowInfo); + +auto queues = getPayload("queues", flowInfo); + +auto versionedFlowSnapshotURI = getPayload("versionedFlowSnapshotURI", flowInfo); + +if (components && queues && versionedFlowSnapshotURI) { + byte = true; + stream->write(byte); + size = components->getNestedPayloads().size(); + stream->write(size); + // write statuses + for (const auto &component : components->g
[GitHub] nifi-minifi-cpp pull request #437: MINIFICPP-558: initial provisioning for C...
Github user arpadboda commented on a diff in the pull request: https://github.com/apache/nifi-minifi-cpp/pull/437#discussion_r233507600 --- Diff: extensions/coap/protocols/CoapC2Protocol.cpp --- @@ -0,0 +1,353 @@ +/** + * + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +#include "CoapC2Protocol.h" +#include "c2/PayloadSerializer.h" + +namespace org { +namespace apache { +namespace nifi { +namespace minifi { +namespace c2 { + +#include "coap_functions.h" +#include "io/BaseStream.h" + +CoapProtocol::CoapProtocol(std::string name, utils::Identifier uuid) +: RESTSender(name, uuid), + require_registration_(false), + logger_(logging::LoggerFactory::getLogger()) { +} + +CoapProtocol::~CoapProtocol() { +} + +void CoapProtocol::initialize(const std::shared_ptr &controller, const std::shared_ptr &configure) { + RESTSender::initialize(controller, configure); + if (configure->get("nifi.c2.coap.connector.service", controller_service_name_)) { +auto service = controller->getControllerService(controller_service_name_); +coap_service_ = std::static_pointer_cast(service); + } else { +logger_->log_info("No CoAP connector configured, so using default service"); +coap_service_ = std::make_shared("cs", configure); +coap_service_->onEnable(); + } +} + +C2Payload CoapProtocol::consumePayload(const std::string &url, const C2Payload &payload, Direction direction, bool async) { + return RESTSender::consumePayload(url, payload, direction, false); +} + +int CoapProtocol::writeAcknowledgement(io::BaseStream *stream, const C2Payload &payload) { + auto ident = payload.getIdentifier(); + auto state = payload.getStatus().getState(); + stream->writeUTF(ident); + uint8_t payloadState = 0; + switch (state) { +case state::UpdateState::NESTED: +case state::UpdateState::INITIATE: +case state::UpdateState::FULLY_APPLIED: +case state::UpdateState::READ_COMPLETE: + payloadState = 0; + break; +case state::UpdateState::NOT_APPLIED: +case state::UpdateState::PARTIALLY_APPLIED: + payloadState = 1; + break; +case state::UpdateState::READ_ERROR: + payloadState = 2; + break; +case state::UpdateState::SET_ERROR: + payloadState = 3; + break; + } + stream->write(&payloadState, 1); + return 0; +} + +int CoapProtocol::writeHeartbeat(io::BaseStream *stream, const C2Payload &payload) { + bool byte; + uint16_t size = 0; + + std::string deviceIdent; + // device identifier + auto deviceInfo = getPayload("deviceInfo", payload); + if (deviceInfo) { +for (const auto &component : deviceInfo->getContent()) { + if (!getString(&component, "identifier", &deviceIdent)) { +break; + } +} + } + stream->writeUTF(deviceIdent, false); + std::string agentIdent; + // agent identifier + auto agentInfo = getPayload("agentInfo", payload); + if (agentInfo) { +for (const auto &component : agentInfo->getContent()) { + if (!getString(&component, "identifier", &agentIdent)) { +break; + } +} + } + if (agentIdent.empty()) { +return -1; + } + stream->writeUTF(agentIdent, false); + + auto flowInfo = getPayload("flowInfo", payload); + + if (flowInfo != nullptr) { + +auto components = getPayload("components", flowInfo); + +auto queues = getPayload("queues", flowInfo); + +auto versionedFlowSnapshotURI = getPayload("versionedFlowSnapshotURI", flowInfo); + +if (components && queues && versionedFlowSnapshotURI) { + byte = true; + stream->write(byte); + size = components->getNestedPayloads().size(); + stream->write(size); + // write statuses + for (const auto &component : components->g
[GitHub] nifi-minifi-cpp pull request #437: MINIFICPP-558: initial provisioning for C...
Github user arpadboda commented on a diff in the pull request: https://github.com/apache/nifi-minifi-cpp/pull/437#discussion_r233941448 --- Diff: extensions/coap/controllerservice/CoapConnector.cpp --- @@ -0,0 +1,189 @@ +/** + * + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "CoapConnector.h" +#include +#include +#include +#include +#include +#include "core/Property.h" +#include "CoapConnector.h" +#include "io/validation.h" +#include "properties/Configure.h" + +namespace org { +namespace apache { +namespace nifi { +namespace minifi { +namespace controllers { + +static core::Property RemoteServer; +static core::Property Port; +static core::Property MaxQueueSize; + +core::Property CoapConnectorService::RemoteServer(core::PropertyBuilder::createProperty("Remote Server")->withDescription("Remote CoAP server")->isRequired(true)->build()); +core::Property CoapConnectorService::Port(core::PropertyBuilder::createProperty("Remote Port")->withDescription("Remote CoAP server port")->isRequired(true)->build()); +core::Property CoapConnectorService::MaxQueueSize(core::PropertyBuilder::createProperty("Max Queue Size")->withDescription("Max queue size for received data ")->isRequired(true)->build()); + +void CoapConnectorService::initialize() { + if (initialized_) +return; + + callback_pointers ptrs; + ptrs.data_received = receiveMessage; + ptrs.received_error = receiveError; + init_coap_api(this, &ptrs); + + std::lock_guard lock(initialization_mutex_); + + ControllerService::initialize(); + + initializeProperties(); + + initialized_ = true; +} + +void CoapConnectorService::onEnable() { + std::string port_str; + if (getProperty(RemoteServer.getName(), host_) && !host_.empty() && getProperty(Port.getName(), port_str) && !port_str.empty()) { +core::Property::StringToInt(port_str, port_); + } else { +// this is the case where we aren't being used in the context of a single controller service. +if (configuration_->get("nifi.c2.agent.coap.host", host_) && configuration_->get("nifi.c2.agent.coap.port", port_str)) { + core::Property::StringToInt(port_str, port_); +} + + } +} + +CoapConnectorService::CoAPMessage CoapConnectorService::sendPayload(uint8_t type, const std::string endpoint, unsigned char *payload, size_t size) { + struct coap_context_t* ctx=NULL; + struct coap_session_t* session=NULL; + + coap_address_t dst_addr, src_addr; + coap_uri_t uri; + uri.host.s = reinterpret_cast(const_cast(host_.c_str())); + uri.host.length = host_.size(); + uri.path.s = reinterpret_cast(const_cast(endpoint.c_str())); + uri.path.length = endpoint.size(); + uri.port = port_; + + fd_set readfds; + coap_pdu_t* request; + unsigned char get_method = 1; + + int res = resolve_address(&uri.host, &dst_addr.addr.sa); + if (res < 0) { +return CoAPMessage(); + } + + dst_addr.size = res; + dst_addr.addr.sin.sin_port = htons(uri.port); + + void *addrptr = NULL; + char port_str[NI_MAXSERV] = "0"; + char node_str[NI_MAXHOST] = ""; + switch (dst_addr.addr.sa.sa_family) { +case AF_INET: + addrptr = &dst_addr.addr.sin.sin_addr; + + /* create context for IPv4 */ + if (!create_session(&ctx, &session, node_str[0] == 0 ? "0.0.0.0" : node_str, port_str, &dst_addr)) { +break; + } else { +return CoAPMessage(); + } +case AF_INET6: + addrptr = &dst_addr.addr.sin6.sin6_addr; + + /* create context for IPv6 */ + if (!create_session(&ctx, &session, node_str[0] == 0 ? "::" : node_str, port_str, &dst_addr)) { +break; + } else { +return CoAPMessage(); + } +default: + ; + } + + // we want to register handlers in the event tha
[GitHub] nifi-minifi-cpp pull request #437: MINIFICPP-558: initial provisioning for C...
Github user arpadboda commented on a diff in the pull request: https://github.com/apache/nifi-minifi-cpp/pull/437#discussion_r233942801 --- Diff: extensions/coap/controllerservice/CoapConnector.cpp --- @@ -0,0 +1,189 @@ +/** + * + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "CoapConnector.h" +#include +#include +#include +#include +#include +#include "core/Property.h" +#include "CoapConnector.h" +#include "io/validation.h" +#include "properties/Configure.h" + +namespace org { +namespace apache { +namespace nifi { +namespace minifi { +namespace controllers { + +static core::Property RemoteServer; +static core::Property Port; +static core::Property MaxQueueSize; + +core::Property CoapConnectorService::RemoteServer(core::PropertyBuilder::createProperty("Remote Server")->withDescription("Remote CoAP server")->isRequired(true)->build()); +core::Property CoapConnectorService::Port(core::PropertyBuilder::createProperty("Remote Port")->withDescription("Remote CoAP server port")->isRequired(true)->build()); +core::Property CoapConnectorService::MaxQueueSize(core::PropertyBuilder::createProperty("Max Queue Size")->withDescription("Max queue size for received data ")->isRequired(true)->build()); + +void CoapConnectorService::initialize() { + if (initialized_) +return; + + callback_pointers ptrs; + ptrs.data_received = receiveMessage; + ptrs.received_error = receiveError; + init_coap_api(this, &ptrs); + + std::lock_guard lock(initialization_mutex_); + + ControllerService::initialize(); + + initializeProperties(); + + initialized_ = true; +} + +void CoapConnectorService::onEnable() { + std::string port_str; + if (getProperty(RemoteServer.getName(), host_) && !host_.empty() && getProperty(Port.getName(), port_str) && !port_str.empty()) { +core::Property::StringToInt(port_str, port_); + } else { +// this is the case where we aren't being used in the context of a single controller service. +if (configuration_->get("nifi.c2.agent.coap.host", host_) && configuration_->get("nifi.c2.agent.coap.port", port_str)) { + core::Property::StringToInt(port_str, port_); +} + + } +} + +CoapConnectorService::CoAPMessage CoapConnectorService::sendPayload(uint8_t type, const std::string endpoint, unsigned char *payload, size_t size) { + struct coap_context_t* ctx=NULL; + struct coap_session_t* session=NULL; + + coap_address_t dst_addr, src_addr; + coap_uri_t uri; + uri.host.s = reinterpret_cast(const_cast(host_.c_str())); --- End diff -- Uri.host is coap_str_const_t, which has "s" as const member, so there is no need of const cast here, just reinterpret cast to "const unsigned char *". ---
[GitHub] nifi-minifi-cpp pull request #437: MINIFICPP-558: initial provisioning for C...
Github user phrocker commented on a diff in the pull request: https://github.com/apache/nifi-minifi-cpp/pull/437#discussion_r233950116 --- Diff: extensions/coap/controllerservice/CoapConnector.cpp --- @@ -0,0 +1,189 @@ +/** + * + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "CoapConnector.h" +#include +#include +#include +#include +#include +#include "core/Property.h" +#include "CoapConnector.h" +#include "io/validation.h" +#include "properties/Configure.h" + +namespace org { +namespace apache { +namespace nifi { +namespace minifi { +namespace controllers { + +static core::Property RemoteServer; +static core::Property Port; +static core::Property MaxQueueSize; + +core::Property CoapConnectorService::RemoteServer(core::PropertyBuilder::createProperty("Remote Server")->withDescription("Remote CoAP server")->isRequired(true)->build()); +core::Property CoapConnectorService::Port(core::PropertyBuilder::createProperty("Remote Port")->withDescription("Remote CoAP server port")->isRequired(true)->build()); +core::Property CoapConnectorService::MaxQueueSize(core::PropertyBuilder::createProperty("Max Queue Size")->withDescription("Max queue size for received data ")->isRequired(true)->build()); + +void CoapConnectorService::initialize() { + if (initialized_) +return; + + callback_pointers ptrs; + ptrs.data_received = receiveMessage; + ptrs.received_error = receiveError; + init_coap_api(this, &ptrs); + + std::lock_guard lock(initialization_mutex_); + + ControllerService::initialize(); + + initializeProperties(); + + initialized_ = true; +} + +void CoapConnectorService::onEnable() { + std::string port_str; + if (getProperty(RemoteServer.getName(), host_) && !host_.empty() && getProperty(Port.getName(), port_str) && !port_str.empty()) { +core::Property::StringToInt(port_str, port_); + } else { +// this is the case where we aren't being used in the context of a single controller service. +if (configuration_->get("nifi.c2.agent.coap.host", host_) && configuration_->get("nifi.c2.agent.coap.port", port_str)) { + core::Property::StringToInt(port_str, port_); +} + + } +} + +CoapConnectorService::CoAPMessage CoapConnectorService::sendPayload(uint8_t type, const std::string endpoint, unsigned char *payload, size_t size) { + struct coap_context_t* ctx=NULL; + struct coap_session_t* session=NULL; + + coap_address_t dst_addr, src_addr; + coap_uri_t uri; + uri.host.s = reinterpret_cast(const_cast(host_.c_str())); --- End diff -- Ah thanks for the diligence. This was changed last minute ( from coap to coap2 ) and I may return back to libcoap where the signature is unsigned char. I posted this early for someone to take a gander from an integration perspective ---
[GitHub] nifi-minifi-cpp pull request #437: MINIFICPP-558: initial provisioning for C...
Github user phrocker commented on a diff in the pull request: https://github.com/apache/nifi-minifi-cpp/pull/437#discussion_r233951286 --- Diff: extensions/coap/controllerservice/CoapConnector.cpp --- @@ -0,0 +1,189 @@ +/** + * + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "CoapConnector.h" +#include +#include +#include +#include +#include +#include "core/Property.h" +#include "CoapConnector.h" +#include "io/validation.h" +#include "properties/Configure.h" + +namespace org { +namespace apache { +namespace nifi { +namespace minifi { +namespace controllers { + +static core::Property RemoteServer; +static core::Property Port; +static core::Property MaxQueueSize; + +core::Property CoapConnectorService::RemoteServer(core::PropertyBuilder::createProperty("Remote Server")->withDescription("Remote CoAP server")->isRequired(true)->build()); +core::Property CoapConnectorService::Port(core::PropertyBuilder::createProperty("Remote Port")->withDescription("Remote CoAP server port")->isRequired(true)->build()); +core::Property CoapConnectorService::MaxQueueSize(core::PropertyBuilder::createProperty("Max Queue Size")->withDescription("Max queue size for received data ")->isRequired(true)->build()); + +void CoapConnectorService::initialize() { + if (initialized_) +return; + + callback_pointers ptrs; + ptrs.data_received = receiveMessage; + ptrs.received_error = receiveError; + init_coap_api(this, &ptrs); + + std::lock_guard lock(initialization_mutex_); + + ControllerService::initialize(); + + initializeProperties(); + + initialized_ = true; +} + +void CoapConnectorService::onEnable() { + std::string port_str; + if (getProperty(RemoteServer.getName(), host_) && !host_.empty() && getProperty(Port.getName(), port_str) && !port_str.empty()) { +core::Property::StringToInt(port_str, port_); + } else { +// this is the case where we aren't being used in the context of a single controller service. +if (configuration_->get("nifi.c2.agent.coap.host", host_) && configuration_->get("nifi.c2.agent.coap.port", port_str)) { + core::Property::StringToInt(port_str, port_); +} + + } +} + +CoapConnectorService::CoAPMessage CoapConnectorService::sendPayload(uint8_t type, const std::string endpoint, unsigned char *payload, size_t size) { + struct coap_context_t* ctx=NULL; + struct coap_session_t* session=NULL; + + coap_address_t dst_addr, src_addr; + coap_uri_t uri; + uri.host.s = reinterpret_cast(const_cast(host_.c_str())); + uri.host.length = host_.size(); + uri.path.s = reinterpret_cast(const_cast(endpoint.c_str())); + uri.path.length = endpoint.size(); + uri.port = port_; + + fd_set readfds; + coap_pdu_t* request; + unsigned char get_method = 1; + + int res = resolve_address(&uri.host, &dst_addr.addr.sa); + if (res < 0) { +return CoAPMessage(); + } + + dst_addr.size = res; + dst_addr.addr.sin.sin_port = htons(uri.port); + + void *addrptr = NULL; + char port_str[NI_MAXSERV] = "0"; + char node_str[NI_MAXHOST] = ""; + switch (dst_addr.addr.sa.sa_family) { +case AF_INET: + addrptr = &dst_addr.addr.sin.sin_addr; + + /* create context for IPv4 */ + if (!create_session(&ctx, &session, node_str[0] == 0 ? "0.0.0.0" : node_str, port_str, &dst_addr)) { +break; + } else { +return CoAPMessage(); + } +case AF_INET6: + addrptr = &dst_addr.addr.sin6.sin6_addr; + + /* create context for IPv6 */ + if (!create_session(&ctx, &session, node_str[0] == 0 ? "::" : node_str, port_str, &dst_addr)) { +break; + } else { +return CoAPMessage(); + } +default: + ; + } + + // we want to register handlers in the event that
[GitHub] nifi-minifi-cpp pull request #437: MINIFICPP-558: initial provisioning for C...
Github user phrocker closed the pull request at: https://github.com/apache/nifi-minifi-cpp/pull/437 ---