http://git-wip-us.apache.org/repos/asf/celix/blob/a129b488/remote_services/remote_service_admin_dfi/private/src/remote_service_admin_dfi.c ---------------------------------------------------------------------- diff --git a/remote_services/remote_service_admin_dfi/private/src/remote_service_admin_dfi.c b/remote_services/remote_service_admin_dfi/private/src/remote_service_admin_dfi.c deleted file mode 100644 index c78cc19..0000000 --- a/remote_services/remote_service_admin_dfi/private/src/remote_service_admin_dfi.c +++ /dev/null @@ -1,747 +0,0 @@ -/** - *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. - */ -/* - * remote_service_admin_impl.c - * - * \date May 21, 2015 - * \author <a href="mailto:d...@celix.apache.org";>Apache Celix Project Team</a> - * \copyright Apache License, Version 2.0 - */ -#include <stdio.h> -#include <stdlib.h> - -#include <arpa/inet.h> -#include <netdb.h> -#include <ifaddrs.h> -#include <string.h> -#include <uuid/uuid.h> -#include <curl/curl.h> - -#include <jansson.h> - -#include "import_registration_dfi.h" -#include "export_registration_dfi.h" -#include "dyn_interface.h" - -#include "remote_service_admin.h" -#include "remote_constants.h" -#include "constants.h" -#include "civetweb.h" - -// defines how often the webserver is restarted (with an increased port number) -#define MAX_NUMBER_OF_RESTARTS 5 - -struct remote_service_admin { - bundle_context_pt context; - log_helper_pt loghelper; - - celix_thread_mutex_t exportedServicesLock; - hash_map_pt exportedServices; - - celix_thread_mutex_t importedServicesLock; - hash_map_pt importedServices; - - char *port; - char *ip; - - struct mg_context *ctx; -}; - -struct post { - const char *readptr; - int size; -}; - -struct get { - char *writeptr; - int size; -}; - -#define OSGI_RSA_REMOTE_PROXY_FACTORY "remote_proxy_factory" -#define OSGI_RSA_REMOTE_PROXY_TIMEOUT "remote_proxy_timeout" - -static const char *data_response_headers = - "HTTP/1.1 200 OK\r\n" - "Cache: no-cache\r\n" - "Content-Type: application/json\r\n" - "\r\n"; - -static const char *no_content_response_headers = - "HTTP/1.1 204 OK\r\n"; - -// TODO do we need to specify a non-Amdatu specific configuration type?! -static const char * const CONFIGURATION_TYPE = "org.amdatu.remote.admin.http"; -static const char * const ENDPOINT_URL = "org.amdatu.remote.admin.http.url"; - -static const char *DEFAULT_PORT = "8888"; -static const char *DEFAULT_IP = "127.0.0.1"; - -static const unsigned int DEFAULT_TIMEOUT = 0; - -static int remoteServiceAdmin_callback(struct mg_connection *conn); -static celix_status_t remoteServiceAdmin_createEndpointDescription(remote_service_admin_pt admin, service_reference_pt reference, char *interface, endpoint_description_pt *description); -static celix_status_t remoteServiceAdmin_send(void *handle, endpoint_description_pt endpointDescription, char *request, char **reply, int* replyStatus); -static celix_status_t remoteServiceAdmin_getIpAdress(char* interface, char** ip); -static size_t remoteServiceAdmin_readCallback(void *ptr, size_t size, size_t nmemb, void *userp); -static size_t remoteServiceAdmin_write(void *contents, size_t size, size_t nmemb, void *userp); -static void remoteServiceAdmin_log(remote_service_admin_pt admin, int level, const char *file, int line, const char *msg, ...); - -celix_status_t remoteServiceAdmin_create(bundle_context_pt context, remote_service_admin_pt *admin) { - celix_status_t status = CELIX_SUCCESS; - - *admin = calloc(1, sizeof(**admin)); - - if (!*admin) { - status = CELIX_ENOMEM; - } else { - unsigned int port_counter = 0; - char *port = NULL; - char *ip = NULL; - char *detectedIp = NULL; - (*admin)->context = context; - (*admin)->exportedServices = hashMap_create(NULL, NULL, NULL, NULL); - (*admin)->importedServices = hashMap_create(NULL, NULL, NULL, NULL); - - celixThreadMutex_create(&(*admin)->exportedServicesLock, NULL); - celixThreadMutex_create(&(*admin)->importedServicesLock, NULL); - - if (logHelper_create(context, &(*admin)->loghelper) == CELIX_SUCCESS) { - logHelper_start((*admin)->loghelper); - dynCommon_logSetup((void *)remoteServiceAdmin_log, *admin, 4); - dynType_logSetup((void *)remoteServiceAdmin_log, *admin, 4); - dynFunction_logSetup((void *)remoteServiceAdmin_log, *admin, 4); - dynInterface_logSetup((void *)remoteServiceAdmin_log, *admin, 4); - } - - bundleContext_getProperty(context, "RSA_PORT", &port); - if (port == NULL) { - port = (char *)DEFAULT_PORT; - } - - bundleContext_getProperty(context, "RSA_IP", &ip); - if (ip == NULL) { - char *interface = NULL; - - bundleContext_getProperty(context, "RSA_INTERFACE", &interface); - if ((interface != NULL) && (remoteServiceAdmin_getIpAdress(interface, &detectedIp) != CELIX_SUCCESS)) { - logHelper_log((*admin)->loghelper, OSGI_LOGSERVICE_WARNING, "RSA: Could not retrieve IP adress for interface %s", interface); - } - - if (ip == NULL) { - remoteServiceAdmin_getIpAdress(NULL, &detectedIp); - } - - ip = detectedIp; - } - - if (ip != NULL) { - logHelper_log((*admin)->loghelper, OSGI_LOGSERVICE_INFO, "RSA: Using %s for service annunciation", ip); - (*admin)->ip = strdup(ip); - } - else { - logHelper_log((*admin)->loghelper, OSGI_LOGSERVICE_WARNING, "RSA: No IP address for service annunciation set. Using %s", DEFAULT_IP); - (*admin)->ip = (char*) DEFAULT_IP; - } - - if (detectedIp != NULL) { - free(detectedIp); - } - - // Prepare callbacks structure. We have only one callback, the rest are NULL. - struct mg_callbacks callbacks; - memset(&callbacks, 0, sizeof(callbacks)); - callbacks.begin_request = remoteServiceAdmin_callback; - - do { - char newPort[10]; - const char *options[] = { "listening_ports", port, NULL}; - - (*admin)->ctx = mg_start(&callbacks, (*admin), options); - - if ((*admin)->ctx != NULL) { - logHelper_log((*admin)->loghelper, OSGI_LOGSERVICE_INFO, "RSA: Start webserver: %s", port); - (*admin)->port = strdup(port); - - } - else { - char* endptr = port; - int currentPort = strtol(port, &endptr, 10); - - errno = 0; - - if (*endptr || errno != 0) { - currentPort = strtol(DEFAULT_PORT, NULL, 10); - } - - port_counter++; - snprintf(&newPort[0], 6, "%d", (currentPort+1)); - - logHelper_log((*admin)->loghelper, OSGI_LOGSERVICE_ERROR, "Error while starting rsa server on port %s - retrying on port %s...", port, newPort); - port = newPort; - } - } while(((*admin)->ctx == NULL) && (port_counter < MAX_NUMBER_OF_RESTARTS)); - - } - return status; -} - - -celix_status_t remoteServiceAdmin_destroy(remote_service_admin_pt *admin) -{ - celix_status_t status = CELIX_SUCCESS; - - free((*admin)->ip); - free((*admin)->port); - free(*admin); - - //TODO destroy exports/imports - - *admin = NULL; - - return status; -} - - -celix_status_t remoteServiceAdmin_stop(remote_service_admin_pt admin) { - celix_status_t status = CELIX_SUCCESS; - - celixThreadMutex_lock(&admin->exportedServicesLock); - - hash_map_iterator_pt iter = hashMapIterator_create(admin->exportedServices); - while (hashMapIterator_hasNext(iter)) { - array_list_pt exports = hashMapIterator_nextValue(iter); - int i; - for (i = 0; i < arrayList_size(exports); i++) { - export_registration_pt export = arrayList_get(exports, i); - if (export != NULL) { - exportRegistration_stop(export); - } - } - } - hashMapIterator_destroy(iter); - celixThreadMutex_unlock(&admin->exportedServicesLock); - - celixThreadMutex_lock(&admin->importedServicesLock); - - iter = hashMapIterator_create(admin->importedServices); - while (hashMapIterator_hasNext(iter)) - { - - hash_map_entry_pt entry = hashMapIterator_nextEntry(iter); - import_registration_pt import = hashMapEntry_getValue(entry); - - if (import != NULL) { - importRegistration_stop(import); - } - } - - hashMapIterator_destroy(iter); - celixThreadMutex_unlock(&admin->importedServicesLock); - - if (admin->ctx != NULL) { - logHelper_log(admin->loghelper, OSGI_LOGSERVICE_INFO, "RSA: Stopping webserver..."); - mg_stop(admin->ctx); - admin->ctx = NULL; - } - - hashMap_destroy(admin->exportedServices, false, false); - hashMap_destroy(admin->importedServices, false, false); - - logHelper_stop(admin->loghelper); - logHelper_destroy(&admin->loghelper); - - return status; -} - -/** - * Request: http://host:port/services/{service}/{request} - */ -//void *remoteServiceAdmin_callback(enum mg_event event, struct mg_connection *conn, const struct mg_request_info *request_info) { - -celix_status_t importRegistration_getFactory(import_registration_pt import, service_factory_pt *factory); - -static int remoteServiceAdmin_callback(struct mg_connection *conn) { - int result = 1; // zero means: let civetweb handle it further, any non-zero value means it is handled by us... - - const struct mg_request_info *request_info = mg_get_request_info(conn); - if (request_info->uri != NULL) { - remote_service_admin_pt rsa = request_info->user_data; - - - if (strncmp(request_info->uri, "/service/", 9) == 0 && strcmp("POST", request_info->request_method) == 0) { - - // uri = /services/myservice/call - const char *uri = request_info->uri; - // rest = myservice/call - - const char *rest = uri+9; - char *interfaceStart = strchr(rest, '/'); - int pos = interfaceStart - rest; - char service[pos+1]; - strncpy(service, rest, pos); - service[pos] = '\0'; - long serviceId = atol(service); - - celixThreadMutex_lock(&rsa->exportedServicesLock); - - //find endpoint - export_registration_pt export = NULL; - hash_map_iterator_pt iter = hashMapIterator_create(rsa->exportedServices); - while (hashMapIterator_hasNext(iter)) { - hash_map_entry_pt entry = hashMapIterator_nextEntry(iter); - array_list_pt exports = hashMapEntry_getValue(entry); - int expIt = 0; - for (expIt = 0; expIt < arrayList_size(exports); expIt++) { - export_registration_pt check = arrayList_get(exports, expIt); - export_reference_pt ref = NULL; - exportRegistration_getExportReference(check, &ref); - endpoint_description_pt checkEndpoint = NULL; - exportReference_getExportedEndpoint(ref, &checkEndpoint); - if (serviceId == checkEndpoint->serviceId) { - export = check; - break; - } - } - } - hashMapIterator_destroy(iter); - - if (export != NULL) { - - uint64_t datalength = request_info->content_length; - char* data = malloc(datalength + 1); - mg_read(conn, data, datalength); - data[datalength] = '\0'; - - char *response = NULL; - int responceLength = 0; - int rc = exportRegistration_call(export, data, -1, &response, &responceLength); - //TODO check rc - - if (response != NULL) { - mg_write(conn, data_response_headers, strlen(data_response_headers)); -// mg_write(conn, no_content_response_headers, strlen(no_content_response_headers)); - printf("writing response '%s'\n", response); - mg_write(conn, response, strlen(response)); -// mg_send_data(conn, response, strlen(response)); -// mg_write_data(conn, response, strlen(response)); - - free(response); - } else { - mg_write(conn, no_content_response_headers, strlen(no_content_response_headers)); - } - result = 1; - - free(data); - } else { - result = 0; - //TODO log warning - } - - celixThreadMutex_unlock(&rsa->exportedServicesLock); - - } - } - - return result; -} - -celix_status_t remoteServiceAdmin_exportService(remote_service_admin_pt admin, char *serviceId, properties_pt properties, array_list_pt *registrations) { - celix_status_t status = CELIX_SUCCESS; - - arrayList_create(registrations); - array_list_pt references = NULL; - service_reference_pt reference = NULL; - char filter [256]; - - snprintf(filter, 256, "(%s=%s)", (char *)OSGI_FRAMEWORK_SERVICE_ID, serviceId); - - status = bundleContext_getServiceReferences(admin->context, NULL, filter, &references); - - logHelper_log(admin->loghelper, OSGI_LOGSERVICE_ERROR, "RSA: exportService called for serviceId %s", serviceId); - - if (status == CELIX_SUCCESS && arrayList_size(references) >= 1) { - reference = arrayList_get(references, 0); - } - - if(references != NULL){ - arrayList_destroy(references); - } - - if (reference == NULL) { - logHelper_log(admin->loghelper, OSGI_LOGSERVICE_ERROR, "ERROR: expected a reference for service id %s.", serviceId); - status = CELIX_ILLEGAL_STATE; - } - - char *exports = NULL; - char *provided = NULL; - if (status == CELIX_SUCCESS) { - serviceReference_getProperty(reference, (char *) OSGI_RSA_SERVICE_EXPORTED_INTERFACES, &exports); - serviceReference_getProperty(reference, (char *) OSGI_FRAMEWORK_OBJECTCLASS, &provided); - - if (exports == NULL || provided == NULL || strcmp(exports, provided) != 0) { - logHelper_log(admin->loghelper, OSGI_LOGSERVICE_WARNING, "RSA: No Services to export."); - status = CELIX_ILLEGAL_STATE; - } else { - logHelper_log(admin->loghelper, OSGI_LOGSERVICE_INFO, "RSA: Export service (%s)", provided); - } - } - - if (status == CELIX_SUCCESS) { - char *interface = provided; - endpoint_description_pt endpoint = NULL; - export_registration_pt registration = NULL; - - remoteServiceAdmin_createEndpointDescription(admin, reference, interface, &endpoint); - printf("RSA: Creating export registration with endpoint pointer %p\n", endpoint); - //TOOD precheck if descriptor exists - status = exportRegistration_create(admin->loghelper, reference, endpoint, admin->context, ®istration); - if (status == CELIX_SUCCESS) { - status = exportRegistration_start(registration); - if (status == CELIX_SUCCESS) { - arrayList_add(*registrations, registration); - } - } - } - - - if (status == CELIX_SUCCESS) { - celixThreadMutex_lock(&admin->exportedServicesLock); - hashMap_put(admin->exportedServices, reference, *registrations); - celixThreadMutex_unlock(&admin->exportedServicesLock); - } - - return status; -} - -celix_status_t remoteServiceAdmin_removeExportedService(export_registration_pt registration) { - celix_status_t status = CELIX_SUCCESS; - //TODO - /* - remote_service_admin_pt admin = registration->rsa; - - celixThreadMutex_lock(&admin->exportedServicesLock); - - hashMap_remove(admin->exportedServices, registration->reference); - //TODO stop ? - - celixThreadMutex_unlock(&admin->exportedServicesLock); - */ - return status; -} - -static celix_status_t remoteServiceAdmin_createEndpointDescription(remote_service_admin_pt admin, service_reference_pt reference, char *interface, endpoint_description_pt *endpoint) { - celix_status_t status = CELIX_SUCCESS; - properties_pt endpointProperties = properties_create(); - - - unsigned int size = 0; - char **keys; - - serviceReference_getPropertyKeys(reference, &keys, &size); - for (int i = 0; i < size; i++) { - char *key = keys[i]; - char *value = NULL; - - if (serviceReference_getProperty(reference, key, &value) == CELIX_SUCCESS - && strcmp(key, (char*) OSGI_RSA_SERVICE_EXPORTED_INTERFACES) != 0 - && strcmp(key, (char*) OSGI_FRAMEWORK_OBJECTCLASS) != 0) { - properties_set(endpointProperties, key, value); - printf("Added property '%s' with value '%s'\n", key, value); - } - } - - hash_map_entry_pt entry = hashMap_getEntry(endpointProperties, (void *) OSGI_FRAMEWORK_SERVICE_ID); - - char* key = hashMapEntry_getKey(entry); - char *serviceId = (char *) hashMap_remove(endpointProperties, (void *) OSGI_FRAMEWORK_SERVICE_ID); - char *uuid = NULL; - - char buf[512]; - snprintf(buf, 512, "/service/%s/%s", serviceId, interface); - - char url[1024]; - snprintf(url, 1024, "http://%s:%s%s";, admin->ip, admin->port, buf); - - uuid_t endpoint_uid; - uuid_generate(endpoint_uid); - char endpoint_uuid[37]; - uuid_unparse_lower(endpoint_uid, endpoint_uuid); - - bundleContext_getProperty(admin->context, OSGI_FRAMEWORK_FRAMEWORK_UUID, &uuid); - properties_set(endpointProperties, (char*) OSGI_RSA_ENDPOINT_FRAMEWORK_UUID, uuid); - properties_set(endpointProperties, (char*) OSGI_FRAMEWORK_OBJECTCLASS, interface); - properties_set(endpointProperties, (char*) OSGI_RSA_ENDPOINT_SERVICE_ID, serviceId); - properties_set(endpointProperties, (char*) OSGI_RSA_ENDPOINT_ID, endpoint_uuid); - properties_set(endpointProperties, (char*) OSGI_RSA_SERVICE_IMPORTED, "true"); - properties_set(endpointProperties, (char*) OSGI_RSA_SERVICE_IMPORTED_CONFIGS, (char*) CONFIGURATION_TYPE); - properties_set(endpointProperties, (char*) ENDPOINT_URL, url); - - - - *endpoint = calloc(1, sizeof(**endpoint)); - if (!*endpoint) { - status = CELIX_ENOMEM; - } else { - (*endpoint)->id = properties_get(endpointProperties, (char*) OSGI_RSA_ENDPOINT_ID); - char *serviceId = NULL; - serviceReference_getProperty(reference, (char*) OSGI_FRAMEWORK_SERVICE_ID, &serviceId); - (*endpoint)->serviceId = strtoull(serviceId, NULL, 0); - (*endpoint)->frameworkUUID = properties_get(endpointProperties, (char*) OSGI_RSA_ENDPOINT_FRAMEWORK_UUID); - (*endpoint)->service = interface; - (*endpoint)->properties = endpointProperties; - } - - free(key); - free(serviceId); - free(keys); - - return status; -} - -static celix_status_t remoteServiceAdmin_getIpAdress(char* interface, char** ip) { - celix_status_t status = CELIX_BUNDLE_EXCEPTION; - - struct ifaddrs *ifaddr, *ifa; - char host[NI_MAXHOST]; - - if (getifaddrs(&ifaddr) != -1) - { - for (ifa = ifaddr; ifa != NULL && status != CELIX_SUCCESS; ifa = ifa->ifa_next) - { - if (ifa->ifa_addr == NULL) - continue; - - if ((getnameinfo(ifa->ifa_addr,sizeof(struct sockaddr_in), host, NI_MAXHOST, NULL, 0, NI_NUMERICHOST) == 0) && (ifa->ifa_addr->sa_family == AF_INET)) { - if (interface == NULL) { - *ip = strdup(host); - status = CELIX_SUCCESS; - } - else if (strcmp(ifa->ifa_name, interface) == 0) { - *ip = strdup(host); - status = CELIX_SUCCESS; - } - } - } - - freeifaddrs(ifaddr); - } - - return status; -} - - -celix_status_t remoteServiceAdmin_destroyEndpointDescription(endpoint_description_pt *description) -{ - celix_status_t status = CELIX_SUCCESS; - - properties_destroy((*description)->properties); - free(*description); - - return status; -} - - -celix_status_t remoteServiceAdmin_getExportedServices(remote_service_admin_pt admin, array_list_pt *services) { - celix_status_t status = CELIX_SUCCESS; - return status; -} - -celix_status_t remoteServiceAdmin_getImportedEndpoints(remote_service_admin_pt admin, array_list_pt *services) { - celix_status_t status = CELIX_SUCCESS; - return status; -} - -celix_status_t remoteServiceAdmin_importService(remote_service_admin_pt admin, endpoint_description_pt endpointDescription, import_registration_pt *out) { - celix_status_t status = CELIX_SUCCESS; - - logHelper_log(admin->loghelper, OSGI_LOGSERVICE_INFO, "RSA: Import service %s", endpointDescription->service); - - const char *objectClass = properties_get(endpointDescription->properties, "objectClass"); - logHelper_log(admin->loghelper, OSGI_LOGSERVICE_INFO, "Registering service factory (proxy) for service '%s'\n", objectClass); - - - import_registration_pt import = NULL; - if (objectClass != NULL) { - status = importRegistration_create(admin->context, endpointDescription, objectClass, &import); - } - if (status == CELIX_SUCCESS) { - importRegistration_setSendFn(import, remoteServiceAdmin_send, admin); - } - - if (status == CELIX_SUCCESS) { - status = importRegistration_start(import); - } - - //celixThreadMutex_lock(&admin->importedServicesLock); - //TODO add to list - //celixThreadMutex_unlock(&admin->importedServicesLock); - - if (status == CELIX_SUCCESS) { - *out = import; - } - - return status; -} - - -celix_status_t remoteServiceAdmin_removeImportedService(remote_service_admin_pt admin, import_registration_pt registration) { - celix_status_t status = CELIX_SUCCESS; - return status; - /* - - endpoint_description_pt endpointDescription = (endpoint_description_pt) registration->endpointDescription; - import_registration_factory_pt registration_factory = NULL; - - celixThreadMutex_lock(&admin->importedServicesLock); - - registration_factory = (import_registration_factory_pt) hashMap_get(admin->importedServices, endpointDescription->service); - - // factory available - if ((registration_factory == NULL) || (registration_factory->trackedFactory == NULL)) - { - logHelper_log(admin->loghelper, OSGI_LOGSERVICE_ERROR, "RSA: Error while retrieving registration factory for imported service %s", endpointDescription->service); - } - else - { - registration_factory->trackedFactory->unregisterProxyService(registration_factory->trackedFactory->factory, endpointDescription); - arrayList_removeElement(registration_factory->registrations, registration); - importRegistration_destroy(registration); - - if (arrayList_isEmpty(registration_factory->registrations)) - { - logHelper_log(admin->loghelper, OSGI_LOGSERVICE_INFO, "RSA: closing proxy."); - - serviceTracker_close(registration_factory->proxyFactoryTracker); - importRegistrationFactory_close(registration_factory); - - hashMap_remove(admin->importedServices, endpointDescription->service); - - importRegistrationFactory_destroy(®istration_factory); - } - } - - celixThreadMutex_unlock(&admin->importedServicesLock); - - return status; - */ -} - - -static celix_status_t remoteServiceAdmin_send(void *handle, endpoint_description_pt endpointDescription, char *request, char **reply, int* replyStatus) { - remote_service_admin_pt rsa = handle; - struct post post; - post.readptr = request; - post.size = strlen(request); - - struct get get; - get.size = 0; - get.writeptr = malloc(1); - - char *serviceUrl = properties_get(endpointDescription->properties, (char*) ENDPOINT_URL); - char url[256]; - snprintf(url, 256, "%s", serviceUrl); - - // assume the default timeout - int timeout = DEFAULT_TIMEOUT; - - char *timeoutStr = NULL; - // Check if the endpoint has a timeout, if so, use it. - timeoutStr = properties_get(endpointDescription->properties, (char*) OSGI_RSA_REMOTE_PROXY_TIMEOUT); - if (timeoutStr == NULL) { - // If not, get the global variable and use that one. - bundleContext_getProperty(rsa->context, (char*) OSGI_RSA_REMOTE_PROXY_TIMEOUT, &timeoutStr); - } - - // Update timeout if a property is used to set it. - if (timeoutStr != NULL) { - timeout = atoi(timeoutStr); - } - - celix_status_t status = CELIX_SUCCESS; - CURL *curl; - CURLcode res; - - curl = curl_easy_init(); - if(!curl) { - status = CELIX_ILLEGAL_STATE; - } else { - curl_easy_setopt(curl, CURLOPT_TIMEOUT, timeout); - curl_easy_setopt(curl, CURLOPT_URL, &url[0]); - curl_easy_setopt(curl, CURLOPT_POST, 1L); - curl_easy_setopt(curl, CURLOPT_READFUNCTION, remoteServiceAdmin_readCallback); - curl_easy_setopt(curl, CURLOPT_READDATA, &post); - curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, remoteServiceAdmin_write); - curl_easy_setopt(curl, CURLOPT_WRITEDATA, (void *)&get); - curl_easy_setopt(curl, CURLOPT_POSTFIELDSIZE, (curl_off_t)post.size); - logHelper_log(rsa->loghelper, OSGI_LOGSERVICE_DEBUG, "RSA: Performing curl post\n"); - res = curl_easy_perform(curl); - - *reply = get.writeptr; - *replyStatus = res; - - curl_easy_cleanup(curl); - } - - return status; -} - -static size_t remoteServiceAdmin_readCallback(void *ptr, size_t size, size_t nmemb, void *userp) { - struct post *post = userp; - - if (post->size) { - *(char *) ptr = post->readptr[0]; - post->readptr++; - post->size--; - return 1; - } - - return 0; -} - -static size_t remoteServiceAdmin_write(void *contents, size_t size, size_t nmemb, void *userp) { - size_t realsize = size * nmemb; - struct get *mem = (struct get *)userp; - - mem->writeptr = realloc(mem->writeptr, mem->size + realsize + 1); - if (mem->writeptr == NULL) { - /* out of memory! */ - printf("not enough memory (realloc returned NULL)"); - exit(EXIT_FAILURE); - } - - memcpy(&(mem->writeptr[mem->size]), contents, realsize); - mem->size += realsize; - mem->writeptr[mem->size] = 0; - - return realsize; -} - - -static void remoteServiceAdmin_log(remote_service_admin_pt admin, int level, const char *file, int line, const char *msg, ...) { - va_list ap; - va_start(ap, msg); - int levels[5] = {0, OSGI_LOGSERVICE_ERROR, OSGI_LOGSERVICE_WARNING, OSGI_LOGSERVICE_INFO, OSGI_LOGSERVICE_DEBUG}; - - char buf1[256]; - snprintf(buf1, 256, "FILE:%s, LINE:%i, MSG:", file, line); - - char buf2[256]; - vsnprintf(buf2, 256, msg, ap); - logHelper_log(admin->loghelper, levels[level], "%s%s", buf1, buf2); -} \ No newline at end of file
http://git-wip-us.apache.org/repos/asf/celix/blob/a129b488/remote_services/remote_service_admin_dfi/rsa/CMakeLists.txt ---------------------------------------------------------------------- diff --git a/remote_services/remote_service_admin_dfi/rsa/CMakeLists.txt b/remote_services/remote_service_admin_dfi/rsa/CMakeLists.txt new file mode 100644 index 0000000..79c9220 --- /dev/null +++ b/remote_services/remote_service_admin_dfi/rsa/CMakeLists.txt @@ -0,0 +1,45 @@ +# 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_directories(private/include) +include_directories("${PROJECT_SOURCE_DIR}/utils/public/include") +include_directories("${PROJECT_SOURCE_DIR}/log_service/public/include") +include_directories("${PROJECT_SOURCE_DIR}/remote_services/utils/private/include") +include_directories("${PROJECT_SOURCE_DIR}/remote_services/remote_service_admin/public/include") +include_directories("${PROJECT_SOURCE_DIR}/remote_services/remote_service_admin_http/private/include") +include_directories("${PROJECT_SOURCE_DIR}/remote_services/remote_service_admin_dfi/dynamic_function_interface") +include_directories("${PROJECT_SOURCE_DIR}/remote_services/examples/calculator_service/public/include") +include_directories("../dynamic_function_interface") + +SET_HEADER(BUNDLE_SYMBOLICNAME "apache_celix_remote_service_admin_dfi") +SET(BUNDLE_VERSION "0.0.1") +SET_HEADERS("Bundle-Name: Apache Celix Remote Service Admin Dynamic Function Interface (DFI)") + +bundle(remote_service_admin_dfi SOURCES + private/src/remote_service_admin_dfi.c + private/src/remote_service_admin_activator.c + private/src/export_registration_dfi.c + private/src/import_registration_dfi.c + + ${PROJECT_SOURCE_DIR}/remote_services/remote_service_admin/private/src/endpoint_description.c + + ${PROJECT_SOURCE_DIR}/remote_services/utils/private/src/civetweb.c + ${PROJECT_SOURCE_DIR}/log_service/public/src/log_helper.c +) +target_link_libraries(remote_service_admin_dfi celix_framework celix_utils ${CURL_LIBRARIES} ${JANSSON_LIBRARIES} dfi) + +install_bundle(remote_service_admin_dfi) http://git-wip-us.apache.org/repos/asf/celix/blob/a129b488/remote_services/remote_service_admin_dfi/rsa/private/include/export_registration_dfi.h ---------------------------------------------------------------------- diff --git a/remote_services/remote_service_admin_dfi/rsa/private/include/export_registration_dfi.h b/remote_services/remote_service_admin_dfi/rsa/private/include/export_registration_dfi.h new file mode 100644 index 0000000..4356646 --- /dev/null +++ b/remote_services/remote_service_admin_dfi/rsa/private/include/export_registration_dfi.h @@ -0,0 +1,21 @@ +/** + * Licensed under Apache License v2. See LICENSE for more information. + */ +#ifndef CELIX_EXPORT_REGISTRATION_DFI_H +#define CELIX_EXPORT_REGISTRATION_DFI_H + + +#include "export_registration.h" +#include "log_helper.h" +#include "endpoint_description.h" + +celix_status_t exportRegistration_create(log_helper_pt helper, service_reference_pt reference, endpoint_description_pt endpoint, bundle_context_pt context, export_registration_pt *registration); +void exportRegistration_destroy(export_registration_pt registration); + +celix_status_t exportRegistration_start(export_registration_pt registration); +celix_status_t exportRegistration_stop(export_registration_pt registration); + +celix_status_t exportRegistration_call(export_registration_pt export, char *data, int datalength, char **response, int *responseLength); + + +#endif //CELIX_EXPORT_REGISTRATION_DFI_H http://git-wip-us.apache.org/repos/asf/celix/blob/a129b488/remote_services/remote_service_admin_dfi/rsa/private/include/import_registration_dfi.h ---------------------------------------------------------------------- diff --git a/remote_services/remote_service_admin_dfi/rsa/private/include/import_registration_dfi.h b/remote_services/remote_service_admin_dfi/rsa/private/include/import_registration_dfi.h new file mode 100644 index 0000000..ec885fd --- /dev/null +++ b/remote_services/remote_service_admin_dfi/rsa/private/include/import_registration_dfi.h @@ -0,0 +1,25 @@ +/** + * Licensed under Apache License v2. See LICENSE for more information. + */ +#ifndef CELIX_IMPORT_REGISTRATION_DFI_H +#define CELIX_IMPORT_REGISTRATION_DFI_H + +#include "import_registration.h" + +#include <celix_errno.h> + +typedef void (*send_func_type)(void *handle, endpoint_description_pt endpointDescription, char *request, char **reply, int* replyStatus); + +celix_status_t importRegistration_create(bundle_context_pt context, endpoint_description_pt description, const char *classObject, import_registration_pt *import); +void importRegistration_destroy(import_registration_pt import); + +celix_status_t importRegistration_setSendFn(import_registration_pt reg, + send_func_type, + void *handle); +celix_status_t importRegistration_start(import_registration_pt import); +celix_status_t importRegistration_stop(import_registration_pt import); + +celix_status_t importRegistration_getService(import_registration_pt import, bundle_pt bundle, service_registration_pt registration, void **service); +celix_status_t importRegistration_ungetService(import_registration_pt import, bundle_pt bundle, service_registration_pt registration, void **service); + +#endif //CELIX_IMPORT_REGISTRATION_DFI_H http://git-wip-us.apache.org/repos/asf/celix/blob/a129b488/remote_services/remote_service_admin_dfi/rsa/private/include/remote_service_admin_http_impl.h ---------------------------------------------------------------------- diff --git a/remote_services/remote_service_admin_dfi/rsa/private/include/remote_service_admin_http_impl.h b/remote_services/remote_service_admin_dfi/rsa/private/include/remote_service_admin_http_impl.h new file mode 100644 index 0000000..65ca83b --- /dev/null +++ b/remote_services/remote_service_admin_dfi/rsa/private/include/remote_service_admin_http_impl.h @@ -0,0 +1,73 @@ +/** + *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. + */ +/* + * remote_service_admin_http_impl.h + * + * \date Sep 30, 2011 + * \author <a href="mailto:d...@celix.apache.org";>Apache Celix Project Team</a> + * \copyright Apache License, Version 2.0 + */ + +#ifndef REMOTE_SERVICE_ADMIN_HTTP_IMPL_H_ +#define REMOTE_SERVICE_ADMIN_HTTP_IMPL_H_ + +#include "remote_service_admin.h" +#include "log_helper.h" +#include "civetweb.h" + +struct remote_service_admin { + bundle_context_pt context; + log_helper_pt loghelper; + + celix_thread_mutex_t exportedServicesLock; + hash_map_pt exportedServices; + + celix_thread_mutex_t importedServicesLock; + hash_map_pt importedServices; + + char *port; + char *ip; + + struct mg_context *ctx; +}; + + +celix_status_t remoteServiceAdmin_create(bundle_context_pt context, remote_service_admin_pt *admin); +celix_status_t remoteServiceAdmin_destroy(remote_service_admin_pt *admin); + +celix_status_t remoteServiceAdmin_stop(remote_service_admin_pt admin); +celix_status_t remoteServiceAdmin_send(remote_service_admin_pt rsa, endpoint_description_pt endpointDescription, char *methodSignature, char **reply, int* replyStatus); + +celix_status_t remoteServiceAdmin_exportService(remote_service_admin_pt admin, char *serviceId, properties_pt properties, array_list_pt *registrations); +celix_status_t remoteServiceAdmin_removeExportedService(export_registration_pt registration); +celix_status_t remoteServiceAdmin_getExportedServices(remote_service_admin_pt admin, array_list_pt *services); +celix_status_t remoteServiceAdmin_getImportedEndpoints(remote_service_admin_pt admin, array_list_pt *services); +celix_status_t remoteServiceAdmin_importService(remote_service_admin_pt admin, endpoint_description_pt endpoint, import_registration_pt *registration); +celix_status_t remoteServiceAdmin_removeImportedService(remote_service_admin_pt admin, import_registration_pt registration); + + +celix_status_t exportReference_getExportedEndpoint(export_reference_pt reference, endpoint_description_pt *endpoint); +celix_status_t exportReference_getExportedService(export_reference_pt reference); + +celix_status_t importReference_getImportedEndpoint(import_reference_pt reference); +celix_status_t importReference_getImportedService(import_reference_pt reference); + +celix_status_t remoteServiceAdmin_destroyEndpointDescription(endpoint_description_pt *description); + +#endif /* REMOTE_SERVICE_ADMIN_HTTP_IMPL_H_ */ http://git-wip-us.apache.org/repos/asf/celix/blob/a129b488/remote_services/remote_service_admin_dfi/rsa/private/src/export_registration_dfi.c ---------------------------------------------------------------------- diff --git a/remote_services/remote_service_admin_dfi/rsa/private/src/export_registration_dfi.c b/remote_services/remote_service_admin_dfi/rsa/private/src/export_registration_dfi.c new file mode 100644 index 0000000..cc181b3 --- /dev/null +++ b/remote_services/remote_service_admin_dfi/rsa/private/src/export_registration_dfi.c @@ -0,0 +1,239 @@ +/** + * Licensed under Apache License v2. See LICENSE for more information. + */ +#include <jansson.h> +#include <dyn_interface.h> +#include <json_serializer.h> +#include <remote_constants.h> +#include <assert.h> +#include "export_registration.h" +#include "export_registration_dfi.h" + +struct export_reference { + endpoint_description_pt endpoint; //owner + service_reference_pt reference; +}; + +struct export_registration { + bundle_context_pt context; + struct export_reference exportReference; + void *service; + dyn_interface_type *intf; //owner + + //TODO add tracker and lock + bool closed; +}; + +typedef void (*gen_func_type)(void); + +struct generic_service_layout { + void *handle; + gen_func_type methods[]; +}; + +celix_status_t exportRegistration_create(log_helper_pt helper, service_reference_pt reference, endpoint_description_pt endpoint, bundle_context_pt context, export_registration_pt *out) { + celix_status_t status = CELIX_SUCCESS; + + export_registration_pt reg = calloc(1, sizeof(*reg)); + + if (reg == NULL) { + status = CELIX_ENOMEM; + } + + if (status == CELIX_SUCCESS) { + reg->context = context; + reg->exportReference.endpoint = endpoint; + reg->exportReference.reference = reference; + reg->closed = false; + } + + char *exports = NULL; + CELIX_DO_IF(status, serviceReference_getProperty(reference, (char *) OSGI_RSA_SERVICE_EXPORTED_INTERFACES, &exports)); + + bundle_pt bundle = NULL; + CELIX_DO_IF(status, serviceReference_getBundle(reference, &bundle)); + + + char *descriptorFile = NULL; + if (status == CELIX_SUCCESS) { + char name[128]; + snprintf(name, 128, "%s.descriptor", exports); + status = bundle_getEntry(bundle, name, &descriptorFile); + logHelper_log(helper, OSGI_LOGSERVICE_DEBUG, "RSA: Found descriptor '%s' for %'s'.", descriptorFile, exports); + } + + if (descriptorFile == NULL) { + logHelper_log(helper, OSGI_LOGSERVICE_ERROR, "RSA: Cannot find descrriptor in bundle for service '%s'", exports); + status = CELIX_ILLEGAL_ARGUMENT; + } + + if (status == CELIX_SUCCESS) { + FILE *df = fopen(descriptorFile, "r"); + if (df != NULL) { + int rc = dynInterface_parse(df, ®->intf); + fclose(df); + if (rc != 0) { + status = CELIX_BUNDLE_EXCEPTION; + logHelper_log(helper, OSGI_LOGSERVICE_WARNING, "RSA: Error parsing service descriptor."); + } + } else { + status = CELIX_BUNDLE_EXCEPTION; + logHelper_log(helper, OSGI_LOGSERVICE_ERROR, "Cannot open descriptor '%s'", descriptorFile); + } + } + + if (status == CELIX_SUCCESS) { + *out = reg; + } else { + logHelper_log(helper, OSGI_LOGSERVICE_ERROR, "Error creating export registration"); + exportRegistration_destroy(reg); + } + + return status; +} + +celix_status_t exportRegistration_call(export_registration_pt export, char *data, int datalength, char **responseOut, int *responseLength) { + int status = CELIX_SUCCESS; + //TODO lock/sema export + + printf("Parsing data: %s\n", data); + json_error_t error; + json_t *js_request = json_loads(data, 0, &error); + json_t *arguments = NULL; + const char *sig; + if (js_request) { + if (json_unpack(js_request, "{s:s}", "m", &sig) != 0) { + printf("RSA: Got error '%s'\n", error.text); + } else { + arguments = json_object_get(js_request, "a"); + } + } else { + printf("RSA: got error '%s' for '%s'\n", error.text, data); + return 0; + } + + printf("RSA: Looking for method %s\n", sig); + struct methods_head *methods = NULL; + dynInterface_methods(export->intf, &methods); + struct method_entry *entry = NULL; + struct method_entry *method = NULL; + TAILQ_FOREACH(entry, methods, entries) { + if (strcmp(sig, entry->id) == 0) { + method = entry; + break; + } + } + + if (method == NULL) { + status = CELIX_ILLEGAL_STATE; + } else { + printf("RSA: found method '%s'\n", entry->id); + } + + if (method != NULL) { + + struct generic_service_layout *serv = export->service; + void *handle = serv->handle; + void (*fp)(void) = serv->methods[method->index]; + + json_t *result = NULL; + status = jsonSerializer_call(method->dynFunc, handle, fp, arguments, &result); + + json_decref(js_request); + + if (status == CELIX_SUCCESS) { + printf("creating payload\n"); + json_t *payload = json_object(); + json_object_set_new(payload, "r", result); + + char *response = json_dumps(payload, JSON_DECODE_ANY); + printf("status ptr is %p. response if '%s'\n", status, response); + + *responseOut = response; + *responseLength = -1; + + json_decref(payload); + } + + ///TODO add more status checks + } + + //TODO unlock/sema export + printf("done export reg call\n"); + return status; +} + +void exportRegistration_destroy(export_registration_pt reg) { + if (reg != NULL) { + if (reg->intf != NULL) { + dyn_interface_type *intf = reg->intf; + reg->intf = NULL; + dynInterface_destroy(intf); + } + + if (reg->exportReference.endpoint != NULL) { + endpoint_description_pt ep = reg->exportReference.endpoint; + reg->exportReference.endpoint = NULL; + endpointDescription_destroy(ep); + } + + free(reg); + } +} + +celix_status_t exportRegistration_start(export_registration_pt reg) { + celix_status_t status = CELIX_SUCCESS; + status = bundleContext_getService(reg->context, reg->exportReference.reference, ®->service); //TODO use tracker + return status; +} + +celix_status_t exportRegistration_stop(export_registration_pt reg) { + celix_status_t status = CELIX_SUCCESS; + status = bundleContext_ungetService(reg->context, reg->exportReference.reference, NULL); + return status; +} + +celix_status_t exportRegistration_close(export_registration_pt reg) { + celix_status_t status = CELIX_SUCCESS; + exportRegistration_stop(reg); + //TODO callback to rsa to remove from list + return status; +} + +celix_status_t exportRegistration_getException(export_registration_pt registration) { + celix_status_t status = CELIX_SUCCESS; + //TODO + return status; +} + +celix_status_t exportRegistration_getExportReference(export_registration_pt registration, export_reference_pt *out) { + celix_status_t status = CELIX_SUCCESS; + export_reference_pt ref = calloc(1, sizeof(*ref)); + if (ref != NULL) { + ref->endpoint = registration->exportReference.endpoint; + ref->reference = registration->exportReference.reference; + } else { + status = CELIX_ENOMEM; + //TODO log + } + + if (status == CELIX_SUCCESS) { + *out = ref; + } + + return status; +} + +celix_status_t exportReference_getExportedEndpoint(export_reference_pt reference, endpoint_description_pt *endpoint) { + celix_status_t status = CELIX_SUCCESS; + *endpoint = reference->endpoint; + return status; +} + +celix_status_t exportReference_getExportedService(export_reference_pt reference) { + celix_status_t status = CELIX_SUCCESS; + return status; +} + + + http://git-wip-us.apache.org/repos/asf/celix/blob/a129b488/remote_services/remote_service_admin_dfi/rsa/private/src/import_registration_dfi.c ---------------------------------------------------------------------- diff --git a/remote_services/remote_service_admin_dfi/rsa/private/src/import_registration_dfi.c b/remote_services/remote_service_admin_dfi/rsa/private/src/import_registration_dfi.c new file mode 100644 index 0000000..20ff61d --- /dev/null +++ b/remote_services/remote_service_admin_dfi/rsa/private/src/import_registration_dfi.c @@ -0,0 +1,318 @@ +#include <stdlib.h> +#include <jansson.h> +#include "json_serializer.h" +#include "dyn_interface.h" +#include "import_registration.h" +#include "import_registration_dfi.h" + +struct import_registration { + bundle_context_pt context; + endpoint_description_pt endpoint; //TODO owner? -> free when destroyed + const char *classObject; //NOTE owned by endpoint + send_func_type send; + void *sendHandle; + + service_factory_pt factory; + service_registration_pt factoryReg; + + hash_map_pt proxies; //key -> bundle, value -> service_proxy +}; + +struct service_proxy { + dyn_interface_type *intf; + void *service; + int count; +}; + +static celix_status_t importRegistration_createProxy(import_registration_pt import, bundle_pt bundle, + struct service_proxy **proxy); +static void importRegistration_proxyFunc(void *userData, void *args[], void *returnVal); +static void importRegistration_destroyProxy(struct service_proxy *proxy); + +celix_status_t importRegistration_create(bundle_context_pt context, endpoint_description_pt endpoint, const char *classObject, import_registration_pt *out) { + celix_status_t status = CELIX_SUCCESS; + import_registration_pt reg = calloc(1, sizeof(*reg)); + + if (reg != NULL) { + reg->factory = calloc(1, sizeof(*reg->factory)); + } + + if (reg != NULL && reg->factory != NULL) { + reg->context = context; + reg->endpoint = endpoint; + reg->classObject = classObject; + reg->proxies = hashMap_create(NULL, NULL, NULL, NULL); + + reg->factory->factory = reg; + reg->factory->getService = (void *)importRegistration_getService; + reg->factory->ungetService = (void *)importRegistration_ungetService; + } else { + status = CELIX_ENOMEM; + } + + if (status == CELIX_SUCCESS) { + printf("IMPORT REGISTRATION IS %p\n", reg); + *out = reg; + } + + return status; +} + + +celix_status_t importRegistration_setSendFn(import_registration_pt reg, + send_func_type send, + void *handle) { + reg->send = send; + reg->sendHandle = handle; + + return CELIX_SUCCESS; +} + +void importRegistration_destroy(import_registration_pt import) { + if (import != NULL) { + if (import->proxies != NULL) { + //TODO destroy proxies + hashMap_destroy(import->proxies, false, false); + import->proxies = NULL; + } + if (import->factory != NULL) { + free(import->factory); + } + free(import); + } +} + +celix_status_t importRegistration_start(import_registration_pt import) { + celix_status_t status = CELIX_SUCCESS; + if (import->factoryReg == NULL && import->factory != NULL) { + status = bundleContext_registerServiceFactory(import->context, (char *)import->classObject, import->factory, NULL /*TODO*/, &import->factoryReg); + } else { + status = CELIX_ILLEGAL_STATE; + } + return status; +} + +celix_status_t importRegistration_stop(import_registration_pt import) { + celix_status_t status = CELIX_SUCCESS; + if (import->factoryReg != NULL) { + serviceRegistration_unregister(import->factoryReg); + } + //TODO unregister every serv instance? + return status; +} + + +celix_status_t importRegistration_getService(import_registration_pt import, bundle_pt bundle, service_registration_pt registration, void **out) { + celix_status_t status = CELIX_SUCCESS; + + /* + module_pt module = NULL; + char *name = NULL; + bundle_getCurrentModule(bundle, &module); + module_getSymbolicName(module, &name); + printf("getting service for bundle '%s'\n", name); + */ + + struct service_proxy *proxy = hashMap_get(import->proxies, bundle); //TODO lock + if (proxy == NULL) { + status = importRegistration_createProxy(import, bundle, &proxy); + if (status == CELIX_SUCCESS) { + hashMap_put(import->proxies, bundle, proxy); //TODO lock + } + } + + if (status == CELIX_SUCCESS) { + proxy->count += 1; + *out = proxy->service; + } + + return status; +} + +static celix_status_t importRegistration_createProxy(import_registration_pt import, bundle_pt bundle, struct service_proxy **out) { + celix_status_t status = CELIX_SUCCESS; + + char *descriptorFile = NULL; + char name[128]; + snprintf(name, 128, "%s.descriptor", import->classObject); + status = bundle_getEntry(bundle, name, &descriptorFile); + if (descriptorFile == NULL) { + printf("Cannot find entry '%s'\n", name); + status = CELIX_ILLEGAL_ARGUMENT; + } else { + printf("Found descriptor at '%s'\n", descriptorFile); + } + + struct service_proxy *proxy = NULL; + if (status == CELIX_SUCCESS) { + proxy = calloc(1, sizeof(*proxy)); + if (proxy == NULL) { + status = CELIX_ENOMEM; + } + } + + if (status == CELIX_SUCCESS) { + FILE *df = fopen(descriptorFile, "r"); + if (df != NULL) { + int rc = dynInterface_parse(df, &proxy->intf); + fclose(df); + if (rc != 0) { + status = CELIX_BUNDLE_EXCEPTION; + } + } + } + + + if (status == CELIX_SUCCESS) { + size_t count = dynInterface_nrOfMethods(proxy->intf); + proxy->service = calloc(1 + count, sizeof(void *)); + if (proxy->service == NULL) { + status = CELIX_ENOMEM; + } + } + + if (status == CELIX_SUCCESS) { + void **serv = proxy->service; + serv[0] = import; + + struct methods_head *list = NULL; + dynInterface_methods(proxy->intf, &list); + struct method_entry *entry = NULL; + void (*fn)(void) = NULL; + int index = 0; + TAILQ_FOREACH(entry, list, entries) { + int rc = dynFunction_createClosure(entry->dynFunc, importRegistration_proxyFunc, entry, &fn); + serv[index + 1] = fn; + index += 1; + + if (rc != 0) { + status = CELIX_BUNDLE_EXCEPTION; + break; + } + } + } + + if (status == CELIX_SUCCESS) { + *out = proxy; + } else { + if (proxy->intf != NULL) { + dynInterface_destroy(proxy->intf); + proxy->intf = NULL; + } + if (proxy->service != NULL) { + free(proxy->service); + proxy->service = NULL; + } + if (proxy != NULL) { + free(proxy); + } + } + + return status; +} + +static void importRegistration_proxyFunc(void *userData, void *args[], void *returnVal) { + int status = CELIX_SUCCESS; + struct method_entry *entry = userData; + import_registration_pt import = *((void **)args[0]); + + printf("Calling remote function '%s'\n", entry->id); + json_t *invoke = json_object(); + json_object_set(invoke, "m", json_string(entry->id)); + + json_t *arguments = NULL; + + status = jsonSerializer_prepareArguments(entry->dynFunc, args, &arguments); + if (status == CELIX_SUCCESS) { + json_object_set_new(invoke, "a", arguments); + } + + char *output = json_dumps(invoke, JSON_DECODE_ANY); + json_decref(invoke); + + printf("Need to send following json '%s'\n", output); + + if (import != NULL && import->send != NULL) { + char *reply = NULL; + int rc = 0; + printf("sending request\n"); + import->send(import->sendHandle, import->endpoint, output, &reply, &rc); + printf("request sended. got reply '%s'\n", reply); + + json_t *replyJson = json_loads(reply, JSON_DECODE_ANY, NULL); //TODO check + json_t *result = json_object_get(replyJson, "r"); //TODO check + status = jsonSerializer_handleReply(entry->dynFunc, NULL, result, args); + json_decref(result); + + + + + if (status == 0) { + printf("done with proxy func\n"); + } + } else { + printf("Error import of import->send is NULL\n"); + } + + //TODO assert double check if return type is native int + int *rVal = returnVal; + *rVal = status; +} + +celix_status_t importRegistration_ungetService(import_registration_pt import, bundle_pt bundle, service_registration_pt registration, void **out) { + celix_status_t status = CELIX_SUCCESS; + return status; + + /* TODO fix. gives segfault in framework shutdown (import->proxies == NULL) + assert(import != NULL); + assert(import->proxies != NULL); + + struct service_proxy *proxy = hashMap_get(import->proxies, bundle); //TODO lock + if (proxy != NULL) { + if (*out == proxy->service) { + proxy->count -= 1; + } else { + status = CELIX_ILLEGAL_ARGUMENT; + } + + if (proxy->count == 0) { + importRegistration_destroyProxy(proxy); + } + } + */ + + return status; +} + +static void importRegistration_destroyProxy(struct service_proxy *proxy) { + //TODO +} + + +celix_status_t importRegistration_close(import_registration_pt registration) { + celix_status_t status = CELIX_SUCCESS; + //TODO + return status; +} + +celix_status_t importRegistration_getException(import_registration_pt registration) { + celix_status_t status = CELIX_SUCCESS; + //TODO + return status; +} + +celix_status_t importRegistration_getImportReference(import_registration_pt registration, import_reference_pt *reference) { + celix_status_t status = CELIX_SUCCESS; + //TODO + return status; +} + +celix_status_t importReference_getImportedEndpoint(import_reference_pt reference) { + celix_status_t status = CELIX_SUCCESS; + return status; +} + +celix_status_t importReference_getImportedService(import_reference_pt reference) { + celix_status_t status = CELIX_SUCCESS; + return status; +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/celix/blob/a129b488/remote_services/remote_service_admin_dfi/rsa/private/src/remote_service_admin_activator.c ---------------------------------------------------------------------- diff --git a/remote_services/remote_service_admin_dfi/rsa/private/src/remote_service_admin_activator.c b/remote_services/remote_service_admin_dfi/rsa/private/src/remote_service_admin_activator.c new file mode 100644 index 0000000..9961a9b --- /dev/null +++ b/remote_services/remote_service_admin_dfi/rsa/private/src/remote_service_admin_activator.c @@ -0,0 +1,122 @@ +/** + *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. + */ +/* + * remote_service_admin_activator.c + * + * \date Sep 30, 2011 + * \author <a href="mailto:d...@celix.apache.org";>Apache Celix Project Team</a> + * \copyright Apache License, Version 2.0 + */ +#include <stdlib.h> + +#include "bundle_activator.h" +#include "service_registration.h" + +#include "remote_service_admin_http_impl.h" +#include "export_registration_dfi.h" +#include "import_registration_dfi.h" + +struct activator { + remote_service_admin_pt admin; + remote_service_admin_service_pt adminService; + service_registration_pt registration; +}; + +celix_status_t bundleActivator_create(bundle_context_pt context, void **userData) { + celix_status_t status = CELIX_SUCCESS; + struct activator *activator; + + activator = calloc(1, sizeof(*activator)); + if (!activator) { + status = CELIX_ENOMEM; + } else { + activator->admin = NULL; + activator->registration = NULL; + + *userData = activator; + } + + return status; +} + +celix_status_t bundleActivator_start(void * userData, bundle_context_pt context) { + celix_status_t status = CELIX_SUCCESS; + struct activator *activator = userData; + remote_service_admin_service_pt remoteServiceAdmin = NULL; + + status = remoteServiceAdmin_create(context, &activator->admin); + if (status == CELIX_SUCCESS) { + remoteServiceAdmin = calloc(1, sizeof(*remoteServiceAdmin)); + if (!remoteServiceAdmin) { + status = CELIX_ENOMEM; + } else { + remoteServiceAdmin->admin = activator->admin; + remoteServiceAdmin->exportService = remoteServiceAdmin_exportService; + + remoteServiceAdmin->getExportedServices = remoteServiceAdmin_getExportedServices; + remoteServiceAdmin->getImportedEndpoints = remoteServiceAdmin_getImportedEndpoints; + remoteServiceAdmin->importService = remoteServiceAdmin_importService; + + remoteServiceAdmin->exportReference_getExportedEndpoint = exportReference_getExportedEndpoint; + remoteServiceAdmin->exportReference_getExportedService = exportReference_getExportedService; + + remoteServiceAdmin->exportRegistration_close = exportRegistration_close; + remoteServiceAdmin->exportRegistration_getException = exportRegistration_getException; + remoteServiceAdmin->exportRegistration_getExportReference = exportRegistration_getExportReference; + + remoteServiceAdmin->importReference_getImportedEndpoint = importReference_getImportedEndpoint; + remoteServiceAdmin->importReference_getImportedService = importReference_getImportedService; + + remoteServiceAdmin->importRegistration_close = remoteServiceAdmin_removeImportedService; + remoteServiceAdmin->importRegistration_getException = importRegistration_getException; + remoteServiceAdmin->importRegistration_getImportReference = importRegistration_getImportReference; + + status = bundleContext_registerService(context, OSGI_RSA_REMOTE_SERVICE_ADMIN, remoteServiceAdmin, NULL, &activator->registration); + activator->adminService = remoteServiceAdmin; + } + } + + return status; +} + +celix_status_t bundleActivator_stop(void * userData, bundle_context_pt context) { + celix_status_t status = CELIX_SUCCESS; + struct activator *activator = userData; + + remoteServiceAdmin_stop(activator->admin); + serviceRegistration_unregister(activator->registration); + activator->registration = NULL; + + remoteServiceAdmin_destroy(&activator->admin); + + free(activator->adminService); + + return status; +} + +celix_status_t bundleActivator_destroy(void * userData, bundle_context_pt context) { + celix_status_t status = CELIX_SUCCESS; + struct activator *activator = userData; + + free(activator); + + return status; +} + + http://git-wip-us.apache.org/repos/asf/celix/blob/a129b488/remote_services/remote_service_admin_dfi/rsa/private/src/remote_service_admin_dfi.c ---------------------------------------------------------------------- diff --git a/remote_services/remote_service_admin_dfi/rsa/private/src/remote_service_admin_dfi.c b/remote_services/remote_service_admin_dfi/rsa/private/src/remote_service_admin_dfi.c new file mode 100644 index 0000000..c78cc19 --- /dev/null +++ b/remote_services/remote_service_admin_dfi/rsa/private/src/remote_service_admin_dfi.c @@ -0,0 +1,747 @@ +/** + *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. + */ +/* + * remote_service_admin_impl.c + * + * \date May 21, 2015 + * \author <a href="mailto:d...@celix.apache.org";>Apache Celix Project Team</a> + * \copyright Apache License, Version 2.0 + */ +#include <stdio.h> +#include <stdlib.h> + +#include <arpa/inet.h> +#include <netdb.h> +#include <ifaddrs.h> +#include <string.h> +#include <uuid/uuid.h> +#include <curl/curl.h> + +#include <jansson.h> + +#include "import_registration_dfi.h" +#include "export_registration_dfi.h" +#include "dyn_interface.h" + +#include "remote_service_admin.h" +#include "remote_constants.h" +#include "constants.h" +#include "civetweb.h" + +// defines how often the webserver is restarted (with an increased port number) +#define MAX_NUMBER_OF_RESTARTS 5 + +struct remote_service_admin { + bundle_context_pt context; + log_helper_pt loghelper; + + celix_thread_mutex_t exportedServicesLock; + hash_map_pt exportedServices; + + celix_thread_mutex_t importedServicesLock; + hash_map_pt importedServices; + + char *port; + char *ip; + + struct mg_context *ctx; +}; + +struct post { + const char *readptr; + int size; +}; + +struct get { + char *writeptr; + int size; +}; + +#define OSGI_RSA_REMOTE_PROXY_FACTORY "remote_proxy_factory" +#define OSGI_RSA_REMOTE_PROXY_TIMEOUT "remote_proxy_timeout" + +static const char *data_response_headers = + "HTTP/1.1 200 OK\r\n" + "Cache: no-cache\r\n" + "Content-Type: application/json\r\n" + "\r\n"; + +static const char *no_content_response_headers = + "HTTP/1.1 204 OK\r\n"; + +// TODO do we need to specify a non-Amdatu specific configuration type?! +static const char * const CONFIGURATION_TYPE = "org.amdatu.remote.admin.http"; +static const char * const ENDPOINT_URL = "org.amdatu.remote.admin.http.url"; + +static const char *DEFAULT_PORT = "8888"; +static const char *DEFAULT_IP = "127.0.0.1"; + +static const unsigned int DEFAULT_TIMEOUT = 0; + +static int remoteServiceAdmin_callback(struct mg_connection *conn); +static celix_status_t remoteServiceAdmin_createEndpointDescription(remote_service_admin_pt admin, service_reference_pt reference, char *interface, endpoint_description_pt *description); +static celix_status_t remoteServiceAdmin_send(void *handle, endpoint_description_pt endpointDescription, char *request, char **reply, int* replyStatus); +static celix_status_t remoteServiceAdmin_getIpAdress(char* interface, char** ip); +static size_t remoteServiceAdmin_readCallback(void *ptr, size_t size, size_t nmemb, void *userp); +static size_t remoteServiceAdmin_write(void *contents, size_t size, size_t nmemb, void *userp); +static void remoteServiceAdmin_log(remote_service_admin_pt admin, int level, const char *file, int line, const char *msg, ...); + +celix_status_t remoteServiceAdmin_create(bundle_context_pt context, remote_service_admin_pt *admin) { + celix_status_t status = CELIX_SUCCESS; + + *admin = calloc(1, sizeof(**admin)); + + if (!*admin) { + status = CELIX_ENOMEM; + } else { + unsigned int port_counter = 0; + char *port = NULL; + char *ip = NULL; + char *detectedIp = NULL; + (*admin)->context = context; + (*admin)->exportedServices = hashMap_create(NULL, NULL, NULL, NULL); + (*admin)->importedServices = hashMap_create(NULL, NULL, NULL, NULL); + + celixThreadMutex_create(&(*admin)->exportedServicesLock, NULL); + celixThreadMutex_create(&(*admin)->importedServicesLock, NULL); + + if (logHelper_create(context, &(*admin)->loghelper) == CELIX_SUCCESS) { + logHelper_start((*admin)->loghelper); + dynCommon_logSetup((void *)remoteServiceAdmin_log, *admin, 4); + dynType_logSetup((void *)remoteServiceAdmin_log, *admin, 4); + dynFunction_logSetup((void *)remoteServiceAdmin_log, *admin, 4); + dynInterface_logSetup((void *)remoteServiceAdmin_log, *admin, 4); + } + + bundleContext_getProperty(context, "RSA_PORT", &port); + if (port == NULL) { + port = (char *)DEFAULT_PORT; + } + + bundleContext_getProperty(context, "RSA_IP", &ip); + if (ip == NULL) { + char *interface = NULL; + + bundleContext_getProperty(context, "RSA_INTERFACE", &interface); + if ((interface != NULL) && (remoteServiceAdmin_getIpAdress(interface, &detectedIp) != CELIX_SUCCESS)) { + logHelper_log((*admin)->loghelper, OSGI_LOGSERVICE_WARNING, "RSA: Could not retrieve IP adress for interface %s", interface); + } + + if (ip == NULL) { + remoteServiceAdmin_getIpAdress(NULL, &detectedIp); + } + + ip = detectedIp; + } + + if (ip != NULL) { + logHelper_log((*admin)->loghelper, OSGI_LOGSERVICE_INFO, "RSA: Using %s for service annunciation", ip); + (*admin)->ip = strdup(ip); + } + else { + logHelper_log((*admin)->loghelper, OSGI_LOGSERVICE_WARNING, "RSA: No IP address for service annunciation set. Using %s", DEFAULT_IP); + (*admin)->ip = (char*) DEFAULT_IP; + } + + if (detectedIp != NULL) { + free(detectedIp); + } + + // Prepare callbacks structure. We have only one callback, the rest are NULL. + struct mg_callbacks callbacks; + memset(&callbacks, 0, sizeof(callbacks)); + callbacks.begin_request = remoteServiceAdmin_callback; + + do { + char newPort[10]; + const char *options[] = { "listening_ports", port, NULL}; + + (*admin)->ctx = mg_start(&callbacks, (*admin), options); + + if ((*admin)->ctx != NULL) { + logHelper_log((*admin)->loghelper, OSGI_LOGSERVICE_INFO, "RSA: Start webserver: %s", port); + (*admin)->port = strdup(port); + + } + else { + char* endptr = port; + int currentPort = strtol(port, &endptr, 10); + + errno = 0; + + if (*endptr || errno != 0) { + currentPort = strtol(DEFAULT_PORT, NULL, 10); + } + + port_counter++; + snprintf(&newPort[0], 6, "%d", (currentPort+1)); + + logHelper_log((*admin)->loghelper, OSGI_LOGSERVICE_ERROR, "Error while starting rsa server on port %s - retrying on port %s...", port, newPort); + port = newPort; + } + } while(((*admin)->ctx == NULL) && (port_counter < MAX_NUMBER_OF_RESTARTS)); + + } + return status; +} + + +celix_status_t remoteServiceAdmin_destroy(remote_service_admin_pt *admin) +{ + celix_status_t status = CELIX_SUCCESS; + + free((*admin)->ip); + free((*admin)->port); + free(*admin); + + //TODO destroy exports/imports + + *admin = NULL; + + return status; +} + + +celix_status_t remoteServiceAdmin_stop(remote_service_admin_pt admin) { + celix_status_t status = CELIX_SUCCESS; + + celixThreadMutex_lock(&admin->exportedServicesLock); + + hash_map_iterator_pt iter = hashMapIterator_create(admin->exportedServices); + while (hashMapIterator_hasNext(iter)) { + array_list_pt exports = hashMapIterator_nextValue(iter); + int i; + for (i = 0; i < arrayList_size(exports); i++) { + export_registration_pt export = arrayList_get(exports, i); + if (export != NULL) { + exportRegistration_stop(export); + } + } + } + hashMapIterator_destroy(iter); + celixThreadMutex_unlock(&admin->exportedServicesLock); + + celixThreadMutex_lock(&admin->importedServicesLock); + + iter = hashMapIterator_create(admin->importedServices); + while (hashMapIterator_hasNext(iter)) + { + + hash_map_entry_pt entry = hashMapIterator_nextEntry(iter); + import_registration_pt import = hashMapEntry_getValue(entry); + + if (import != NULL) { + importRegistration_stop(import); + } + } + + hashMapIterator_destroy(iter); + celixThreadMutex_unlock(&admin->importedServicesLock); + + if (admin->ctx != NULL) { + logHelper_log(admin->loghelper, OSGI_LOGSERVICE_INFO, "RSA: Stopping webserver..."); + mg_stop(admin->ctx); + admin->ctx = NULL; + } + + hashMap_destroy(admin->exportedServices, false, false); + hashMap_destroy(admin->importedServices, false, false); + + logHelper_stop(admin->loghelper); + logHelper_destroy(&admin->loghelper); + + return status; +} + +/** + * Request: http://host:port/services/{service}/{request} + */ +//void *remoteServiceAdmin_callback(enum mg_event event, struct mg_connection *conn, const struct mg_request_info *request_info) { + +celix_status_t importRegistration_getFactory(import_registration_pt import, service_factory_pt *factory); + +static int remoteServiceAdmin_callback(struct mg_connection *conn) { + int result = 1; // zero means: let civetweb handle it further, any non-zero value means it is handled by us... + + const struct mg_request_info *request_info = mg_get_request_info(conn); + if (request_info->uri != NULL) { + remote_service_admin_pt rsa = request_info->user_data; + + + if (strncmp(request_info->uri, "/service/", 9) == 0 && strcmp("POST", request_info->request_method) == 0) { + + // uri = /services/myservice/call + const char *uri = request_info->uri; + // rest = myservice/call + + const char *rest = uri+9; + char *interfaceStart = strchr(rest, '/'); + int pos = interfaceStart - rest; + char service[pos+1]; + strncpy(service, rest, pos); + service[pos] = '\0'; + long serviceId = atol(service); + + celixThreadMutex_lock(&rsa->exportedServicesLock); + + //find endpoint + export_registration_pt export = NULL; + hash_map_iterator_pt iter = hashMapIterator_create(rsa->exportedServices); + while (hashMapIterator_hasNext(iter)) { + hash_map_entry_pt entry = hashMapIterator_nextEntry(iter); + array_list_pt exports = hashMapEntry_getValue(entry); + int expIt = 0; + for (expIt = 0; expIt < arrayList_size(exports); expIt++) { + export_registration_pt check = arrayList_get(exports, expIt); + export_reference_pt ref = NULL; + exportRegistration_getExportReference(check, &ref); + endpoint_description_pt checkEndpoint = NULL; + exportReference_getExportedEndpoint(ref, &checkEndpoint); + if (serviceId == checkEndpoint->serviceId) { + export = check; + break; + } + } + } + hashMapIterator_destroy(iter); + + if (export != NULL) { + + uint64_t datalength = request_info->content_length; + char* data = malloc(datalength + 1); + mg_read(conn, data, datalength); + data[datalength] = '\0'; + + char *response = NULL; + int responceLength = 0; + int rc = exportRegistration_call(export, data, -1, &response, &responceLength); + //TODO check rc + + if (response != NULL) { + mg_write(conn, data_response_headers, strlen(data_response_headers)); +// mg_write(conn, no_content_response_headers, strlen(no_content_response_headers)); + printf("writing response '%s'\n", response); + mg_write(conn, response, strlen(response)); +// mg_send_data(conn, response, strlen(response)); +// mg_write_data(conn, response, strlen(response)); + + free(response); + } else { + mg_write(conn, no_content_response_headers, strlen(no_content_response_headers)); + } + result = 1; + + free(data); + } else { + result = 0; + //TODO log warning + } + + celixThreadMutex_unlock(&rsa->exportedServicesLock); + + } + } + + return result; +} + +celix_status_t remoteServiceAdmin_exportService(remote_service_admin_pt admin, char *serviceId, properties_pt properties, array_list_pt *registrations) { + celix_status_t status = CELIX_SUCCESS; + + arrayList_create(registrations); + array_list_pt references = NULL; + service_reference_pt reference = NULL; + char filter [256]; + + snprintf(filter, 256, "(%s=%s)", (char *)OSGI_FRAMEWORK_SERVICE_ID, serviceId); + + status = bundleContext_getServiceReferences(admin->context, NULL, filter, &references); + + logHelper_log(admin->loghelper, OSGI_LOGSERVICE_ERROR, "RSA: exportService called for serviceId %s", serviceId); + + if (status == CELIX_SUCCESS && arrayList_size(references) >= 1) { + reference = arrayList_get(references, 0); + } + + if(references != NULL){ + arrayList_destroy(references); + } + + if (reference == NULL) { + logHelper_log(admin->loghelper, OSGI_LOGSERVICE_ERROR, "ERROR: expected a reference for service id %s.", serviceId); + status = CELIX_ILLEGAL_STATE; + } + + char *exports = NULL; + char *provided = NULL; + if (status == CELIX_SUCCESS) { + serviceReference_getProperty(reference, (char *) OSGI_RSA_SERVICE_EXPORTED_INTERFACES, &exports); + serviceReference_getProperty(reference, (char *) OSGI_FRAMEWORK_OBJECTCLASS, &provided); + + if (exports == NULL || provided == NULL || strcmp(exports, provided) != 0) { + logHelper_log(admin->loghelper, OSGI_LOGSERVICE_WARNING, "RSA: No Services to export."); + status = CELIX_ILLEGAL_STATE; + } else { + logHelper_log(admin->loghelper, OSGI_LOGSERVICE_INFO, "RSA: Export service (%s)", provided); + } + } + + if (status == CELIX_SUCCESS) { + char *interface = provided; + endpoint_description_pt endpoint = NULL; + export_registration_pt registration = NULL; + + remoteServiceAdmin_createEndpointDescription(admin, reference, interface, &endpoint); + printf("RSA: Creating export registration with endpoint pointer %p\n", endpoint); + //TOOD precheck if descriptor exists + status = exportRegistration_create(admin->loghelper, reference, endpoint, admin->context, ®istration); + if (status == CELIX_SUCCESS) { + status = exportRegistration_start(registration); + if (status == CELIX_SUCCESS) { + arrayList_add(*registrations, registration); + } + } + } + + + if (status == CELIX_SUCCESS) { + celixThreadMutex_lock(&admin->exportedServicesLock); + hashMap_put(admin->exportedServices, reference, *registrations); + celixThreadMutex_unlock(&admin->exportedServicesLock); + } + + return status; +} + +celix_status_t remoteServiceAdmin_removeExportedService(export_registration_pt registration) { + celix_status_t status = CELIX_SUCCESS; + //TODO + /* + remote_service_admin_pt admin = registration->rsa; + + celixThreadMutex_lock(&admin->exportedServicesLock); + + hashMap_remove(admin->exportedServices, registration->reference); + //TODO stop ? + + celixThreadMutex_unlock(&admin->exportedServicesLock); + */ + return status; +} + +static celix_status_t remoteServiceAdmin_createEndpointDescription(remote_service_admin_pt admin, service_reference_pt reference, char *interface, endpoint_description_pt *endpoint) { + celix_status_t status = CELIX_SUCCESS; + properties_pt endpointProperties = properties_create(); + + + unsigned int size = 0; + char **keys; + + serviceReference_getPropertyKeys(reference, &keys, &size); + for (int i = 0; i < size; i++) { + char *key = keys[i]; + char *value = NULL; + + if (serviceReference_getProperty(reference, key, &value) == CELIX_SUCCESS + && strcmp(key, (char*) OSGI_RSA_SERVICE_EXPORTED_INTERFACES) != 0 + && strcmp(key, (char*) OSGI_FRAMEWORK_OBJECTCLASS) != 0) { + properties_set(endpointProperties, key, value); + printf("Added property '%s' with value '%s'\n", key, value); + } + } + + hash_map_entry_pt entry = hashMap_getEntry(endpointProperties, (void *) OSGI_FRAMEWORK_SERVICE_ID); + + char* key = hashMapEntry_getKey(entry); + char *serviceId = (char *) hashMap_remove(endpointProperties, (void *) OSGI_FRAMEWORK_SERVICE_ID); + char *uuid = NULL; + + char buf[512]; + snprintf(buf, 512, "/service/%s/%s", serviceId, interface); + + char url[1024]; + snprintf(url, 1024, "http://%s:%s%s";, admin->ip, admin->port, buf); + + uuid_t endpoint_uid; + uuid_generate(endpoint_uid); + char endpoint_uuid[37]; + uuid_unparse_lower(endpoint_uid, endpoint_uuid); + + bundleContext_getProperty(admin->context, OSGI_FRAMEWORK_FRAMEWORK_UUID, &uuid); + properties_set(endpointProperties, (char*) OSGI_RSA_ENDPOINT_FRAMEWORK_UUID, uuid); + properties_set(endpointProperties, (char*) OSGI_FRAMEWORK_OBJECTCLASS, interface); + properties_set(endpointProperties, (char*) OSGI_RSA_ENDPOINT_SERVICE_ID, serviceId); + properties_set(endpointProperties, (char*) OSGI_RSA_ENDPOINT_ID, endpoint_uuid); + properties_set(endpointProperties, (char*) OSGI_RSA_SERVICE_IMPORTED, "true"); + properties_set(endpointProperties, (char*) OSGI_RSA_SERVICE_IMPORTED_CONFIGS, (char*) CONFIGURATION_TYPE); + properties_set(endpointProperties, (char*) ENDPOINT_URL, url); + + + + *endpoint = calloc(1, sizeof(**endpoint)); + if (!*endpoint) { + status = CELIX_ENOMEM; + } else { + (*endpoint)->id = properties_get(endpointProperties, (char*) OSGI_RSA_ENDPOINT_ID); + char *serviceId = NULL; + serviceReference_getProperty(reference, (char*) OSGI_FRAMEWORK_SERVICE_ID, &serviceId); + (*endpoint)->serviceId = strtoull(serviceId, NULL, 0); + (*endpoint)->frameworkUUID = properties_get(endpointProperties, (char*) OSGI_RSA_ENDPOINT_FRAMEWORK_UUID); + (*endpoint)->service = interface; + (*endpoint)->properties = endpointProperties; + } + + free(key); + free(serviceId); + free(keys); + + return status; +} + +static celix_status_t remoteServiceAdmin_getIpAdress(char* interface, char** ip) { + celix_status_t status = CELIX_BUNDLE_EXCEPTION; + + struct ifaddrs *ifaddr, *ifa; + char host[NI_MAXHOST]; + + if (getifaddrs(&ifaddr) != -1) + { + for (ifa = ifaddr; ifa != NULL && status != CELIX_SUCCESS; ifa = ifa->ifa_next) + { + if (ifa->ifa_addr == NULL) + continue; + + if ((getnameinfo(ifa->ifa_addr,sizeof(struct sockaddr_in), host, NI_MAXHOST, NULL, 0, NI_NUMERICHOST) == 0) && (ifa->ifa_addr->sa_family == AF_INET)) { + if (interface == NULL) { + *ip = strdup(host); + status = CELIX_SUCCESS; + } + else if (strcmp(ifa->ifa_name, interface) == 0) { + *ip = strdup(host); + status = CELIX_SUCCESS; + } + } + } + + freeifaddrs(ifaddr); + } + + return status; +} + + +celix_status_t remoteServiceAdmin_destroyEndpointDescription(endpoint_description_pt *description) +{ + celix_status_t status = CELIX_SUCCESS; + + properties_destroy((*description)->properties); + free(*description); + + return status; +} + + +celix_status_t remoteServiceAdmin_getExportedServices(remote_service_admin_pt admin, array_list_pt *services) { + celix_status_t status = CELIX_SUCCESS; + return status; +} + +celix_status_t remoteServiceAdmin_getImportedEndpoints(remote_service_admin_pt admin, array_list_pt *services) { + celix_status_t status = CELIX_SUCCESS; + return status; +} + +celix_status_t remoteServiceAdmin_importService(remote_service_admin_pt admin, endpoint_description_pt endpointDescription, import_registration_pt *out) { + celix_status_t status = CELIX_SUCCESS; + + logHelper_log(admin->loghelper, OSGI_LOGSERVICE_INFO, "RSA: Import service %s", endpointDescription->service); + + const char *objectClass = properties_get(endpointDescription->properties, "objectClass"); + logHelper_log(admin->loghelper, OSGI_LOGSERVICE_INFO, "Registering service factory (proxy) for service '%s'\n", objectClass); + + + import_registration_pt import = NULL; + if (objectClass != NULL) { + status = importRegistration_create(admin->context, endpointDescription, objectClass, &import); + } + if (status == CELIX_SUCCESS) { + importRegistration_setSendFn(import, remoteServiceAdmin_send, admin); + } + + if (status == CELIX_SUCCESS) { + status = importRegistration_start(import); + } + + //celixThreadMutex_lock(&admin->importedServicesLock); + //TODO add to list + //celixThreadMutex_unlock(&admin->importedServicesLock); + + if (status == CELIX_SUCCESS) { + *out = import; + } + + return status; +} + + +celix_status_t remoteServiceAdmin_removeImportedService(remote_service_admin_pt admin, import_registration_pt registration) { + celix_status_t status = CELIX_SUCCESS; + return status; + /* + + endpoint_description_pt endpointDescription = (endpoint_description_pt) registration->endpointDescription; + import_registration_factory_pt registration_factory = NULL; + + celixThreadMutex_lock(&admin->importedServicesLock); + + registration_factory = (import_registration_factory_pt) hashMap_get(admin->importedServices, endpointDescription->service); + + // factory available + if ((registration_factory == NULL) || (registration_factory->trackedFactory == NULL)) + { + logHelper_log(admin->loghelper, OSGI_LOGSERVICE_ERROR, "RSA: Error while retrieving registration factory for imported service %s", endpointDescription->service); + } + else + { + registration_factory->trackedFactory->unregisterProxyService(registration_factory->trackedFactory->factory, endpointDescription); + arrayList_removeElement(registration_factory->registrations, registration); + importRegistration_destroy(registration); + + if (arrayList_isEmpty(registration_factory->registrations)) + { + logHelper_log(admin->loghelper, OSGI_LOGSERVICE_INFO, "RSA: closing proxy."); + + serviceTracker_close(registration_factory->proxyFactoryTracker); + importRegistrationFactory_close(registration_factory); + + hashMap_remove(admin->importedServices, endpointDescription->service); + + importRegistrationFactory_destroy(®istration_factory); + } + } + + celixThreadMutex_unlock(&admin->importedServicesLock); + + return status; + */ +} + + +static celix_status_t remoteServiceAdmin_send(void *handle, endpoint_description_pt endpointDescription, char *request, char **reply, int* replyStatus) { + remote_service_admin_pt rsa = handle; + struct post post; + post.readptr = request; + post.size = strlen(request); + + struct get get; + get.size = 0; + get.writeptr = malloc(1); + + char *serviceUrl = properties_get(endpointDescription->properties, (char*) ENDPOINT_URL); + char url[256]; + snprintf(url, 256, "%s", serviceUrl); + + // assume the default timeout + int timeout = DEFAULT_TIMEOUT; + + char *timeoutStr = NULL; + // Check if the endpoint has a timeout, if so, use it. + timeoutStr = properties_get(endpointDescription->properties, (char*) OSGI_RSA_REMOTE_PROXY_TIMEOUT); + if (timeoutStr == NULL) { + // If not, get the global variable and use that one. + bundleContext_getProperty(rsa->context, (char*) OSGI_RSA_REMOTE_PROXY_TIMEOUT, &timeoutStr); + } + + // Update timeout if a property is used to set it. + if (timeoutStr != NULL) { + timeout = atoi(timeoutStr); + } + + celix_status_t status = CELIX_SUCCESS; + CURL *curl; + CURLcode res; + + curl = curl_easy_init(); + if(!curl) { + status = CELIX_ILLEGAL_STATE; + } else { + curl_easy_setopt(curl, CURLOPT_TIMEOUT, timeout); + curl_easy_setopt(curl, CURLOPT_URL, &url[0]); + curl_easy_setopt(curl, CURLOPT_POST, 1L); + curl_easy_setopt(curl, CURLOPT_READFUNCTION, remoteServiceAdmin_readCallback); + curl_easy_setopt(curl, CURLOPT_READDATA, &post); + curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, remoteServiceAdmin_write); + curl_easy_setopt(curl, CURLOPT_WRITEDATA, (void *)&get); + curl_easy_setopt(curl, CURLOPT_POSTFIELDSIZE, (curl_off_t)post.size); + logHelper_log(rsa->loghelper, OSGI_LOGSERVICE_DEBUG, "RSA: Performing curl post\n"); + res = curl_easy_perform(curl); + + *reply = get.writeptr; + *replyStatus = res; + + curl_easy_cleanup(curl); + } + + return status; +} + +static size_t remoteServiceAdmin_readCallback(void *ptr, size_t size, size_t nmemb, void *userp) { + struct post *post = userp; + + if (post->size) { + *(char *) ptr = post->readptr[0]; + post->readptr++; + post->size--; + return 1; + } + + return 0; +} + +static size_t remoteServiceAdmin_write(void *contents, size_t size, size_t nmemb, void *userp) { + size_t realsize = size * nmemb; + struct get *mem = (struct get *)userp; + + mem->writeptr = realloc(mem->writeptr, mem->size + realsize + 1); + if (mem->writeptr == NULL) { + /* out of memory! */ + printf("not enough memory (realloc returned NULL)"); + exit(EXIT_FAILURE); + } + + memcpy(&(mem->writeptr[mem->size]), contents, realsize); + mem->size += realsize; + mem->writeptr[mem->size] = 0; + + return realsize; +} + + +static void remoteServiceAdmin_log(remote_service_admin_pt admin, int level, const char *file, int line, const char *msg, ...) { + va_list ap; + va_start(ap, msg); + int levels[5] = {0, OSGI_LOGSERVICE_ERROR, OSGI_LOGSERVICE_WARNING, OSGI_LOGSERVICE_INFO, OSGI_LOGSERVICE_DEBUG}; + + char buf1[256]; + snprintf(buf1, 256, "FILE:%s, LINE:%i, MSG:", file, line); + + char buf2[256]; + vsnprintf(buf2, 256, msg, ap); + logHelper_log(admin->loghelper, levels[level], "%s%s", buf1, buf2); +} \ No newline at end of file
