File content as of revision 1:0ddbe2d3319c:

/*
 * Copyright (c) 2015 ARM Limited. All rights reserved.
 * SPDX-License-Identifier: Apache-2.0
 * Licensed under the Apache License, Version 2.0 (the License); you may
 * not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an AS IS 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 "mbed-client-classic/m2mconnectionhandlerpimpl.h"
#include "mbed-client/m2mconnectionobserver.h"
#include "mbed-client/m2mconstants.h"
#include "mbed-client/m2msecurity.h"
#include "mbed-client/m2mconnectionhandler.h"

#include "Socket/TCPSocketConnection.h"
#include "Socket/UDPSocket.h"
#include "threadwrapper.h"
#include "mbed_error.h"


M2MConnectionHandlerPimpl::M2MConnectionHandlerPimpl(M2MConnectionHandler* base, M2MConnectionObserver &observer,
                                                     M2MConnectionSecurity* sec,
                                                     M2MInterface::BindingMode mode,
                                                     M2MInterface::NetworkStack stack)
:_base(base),
 _observer(observer),
 _security_impl(sec),
 _use_secure_connection(false),
 _binding_mode(mode),
 _network_stack(stack),
 _resolved(true),
 _is_handshaking(false),
 _listening(false),
 _listen_thread(0),
 _recv_thread(0),
 _send_thread(0)
{
    memset(&_address_buffer, 0, sizeof _address_buffer);
    memset(&_address, 0, sizeof _address);
    _address._address = _address_buffer;

    if (_network_stack != M2MInterface::LwIP_IPv4) {
        error("ConnectionHandler: Unsupported network stack, only IPv4 is currently supported");
    }
    
    if (_binding_mode == M2MInterface::TCP ||
        _binding_mode == M2MInterface::TCP_QUEUE) {
        error("ConnectionHandler: Unsupported binding mode, only UDP based modes are currently supported");
    }
    
    _running = true;
    _recv_thread = rtos::create_thread<
        M2MConnectionHandlerPimpl,
        &M2MConnectionHandlerPimpl::recv_handler>(this, osPriorityAboveNormal);
    _send_thread = rtos::create_thread<
        M2MConnectionHandlerPimpl,
        &M2MConnectionHandlerPimpl::send_handler>(this);
    _listen_thread = rtos::create_thread<
        M2MConnectionHandlerPimpl,
        &M2MConnectionHandlerPimpl::listen_handler>(this);    
}

M2MConnectionHandlerPimpl::~M2MConnectionHandlerPimpl()
{
    _listening = false;
    _running = false;
    
    if (_listen_thread) {
        delete _listen_thread;
        _listen_thread = 0;
    }
    
    if (_recv_thread) {
        delete _recv_thread;
        _recv_thread = 0;
    }
    
    if (_send_thread) {
        delete _send_thread;
        _send_thread = 0;
    }
    
    delete _security_impl;
}

bool M2MConnectionHandlerPimpl::bind_connection(const uint16_t listen_port)
{
    return !(_socket.bind(listen_port) < 0);
}

bool M2MConnectionHandlerPimpl::resolve_server_address(const String& server_address,
                                                       const uint16_t server_port,
                                                       M2MConnectionObserver::ServerType server_type,
                                                       const M2MSecurity* security)
{
    if (_endpoint.set_address(server_address.c_str(), server_port) < 0) {
        return false;
    }

    inet_aton(_endpoint.get_address(), _address._address);
    _address._port = _endpoint.get_port();
    _address._length = 4;
    _address._stack = _network_stack;

    if (security) {
        if (security->resource_value_int(M2MSecurity::SecurityMode) == M2MSecurity::Certificate ||
            security->resource_value_int(M2MSecurity::SecurityMode) == M2MSecurity::Psk) {
            if (_security_impl != NULL) {
                _security_impl->reset();
                _security_impl->init(security);
                
                if (_security_impl->connect(_base) < 0) {
                    return false;
                }
                
                _use_secure_connection = true;
            }
        }
    }
    
    _observer.address_ready(_address,
                            server_type,
                            _address._port);
    
    return true;
}


void M2MConnectionHandlerPimpl::recv_handler()
{
    while (_running) {
        Endpoint recv_endpoint;
        int size = _socket.receiveFrom(recv_endpoint,
            (char*)_recv_buffer, sizeof _recv_buffer);
    
        if (size > 0) {
            _recv_queue.put(new std::string(_recv_buffer, _recv_buffer+size));
        } else {
            rtos::Thread::wait(1000);
        }
    }
}

void M2MConnectionHandlerPimpl::send_handler()
{    
    while (_running) {
        osEvent e = _send_queue.get();
        if (e.status == osEventMessage) {
            std::string *packet = (std::string*)e.value.p;
            int size = _socket.sendTo(_endpoint, (char*)packet->data(), packet->size());
            delete packet;
        } else {
            rtos::Thread::wait(1000);
        }
    }
}

bool M2MConnectionHandlerPimpl::send_data(uint8_t *data,
                                          uint16_t data_len,
                                          sn_nsdl_addr_s *address)
{
    if (address == NULL || data == NULL) {
        return false;
    }
    
    if (_use_secure_connection) {
        if (_security_impl->send_message(data, data_len) < 0) {
            return false;
        }
    } else {
        if (send_to_socket(data, data_len) < 0) {
            return false;
        }
    }
    
    _observer.data_sent();
    return true;
}

bool M2MConnectionHandlerPimpl::start_listening_for_data()
{
    _listening = true;
    return true;
}

void M2MConnectionHandlerPimpl::stop_listening()
{
    _listening = false;
}

void M2MConnectionHandlerPimpl::listen_handler()
{
    while (_running) {
        if (_listening) {
            memset(_listen_buffer, 0, sizeof _listen_buffer);
            int size;
            
            if (_use_secure_connection) {
                size = _security_impl->read(_listen_buffer, sizeof _listen_buffer);
            } else {
                size = receive_from_socket(_listen_buffer, sizeof _listen_buffer);
            }
            
            if (size > 0) {
                _observer.data_available((uint8_t*)_listen_buffer, size, _address);
            } else if (size != 0) {
                _listening = false;
                _observer.socket_error(2);
            }
        } else {
            rtos::Thread::wait(1000);
        }
    }
}


int M2MConnectionHandlerPimpl::send_to_socket(const unsigned char *buf, size_t len)
{
    if (_send_queue.put(new std::string(buf, buf+len)) != osOK) {
        return M2MConnectionHandler::CONNECTION_ERROR_WANTS_WRITE;
    } else {
        return len;
    }
}

int M2MConnectionHandlerPimpl::receive_from_socket(unsigned char *buf, size_t len)
{
    osEvent e = _recv_queue.get();
    if (e.status == osEventMessage) {
        std::string *packet = (std::string*)e.value.p;
        int size = packet->size();
        
        if (size <= len) {
            memcpy(buf, packet->data(), size);
            delete packet;
            return size;
        }
    }
    
    return M2MConnectionHandler::CONNECTION_ERROR_WANTS_READ;
}

void M2MConnectionHandlerPimpl::handle_connection_error(int /*error*/)
{
    _observer.socket_error(4);
}