celix_status_t discovery_destroy(discovery_pt discovery) {
	celix_status_t status = CELIX_SUCCESS;

	discovery->context = NULL;
	discovery->poller = NULL;
	discovery->server = NULL;

	celixThreadMutex_lock(&discovery->discoveredServicesMutex);

	hashMap_destroy(discovery->discoveredServices, false, false);
	discovery->discoveredServices = NULL;

	celixThreadMutex_unlock(&discovery->discoveredServicesMutex);

	celixThreadMutex_destroy(&discovery->discoveredServicesMutex);

	celixThreadMutex_lock(&discovery->listenerReferencesMutex);

	hashMap_destroy(discovery->listenerReferences, false, false);
	discovery->listenerReferences = NULL;

	celixThreadMutex_unlock(&discovery->listenerReferencesMutex);

	celixThreadMutex_destroy(&discovery->listenerReferencesMutex);

	logHelper_destroy(&discovery->loghelper);

	free(discovery);

	return status;
}

celix_status_t wiringAdmin_destroy(wiring_admin_pt* admin) {
    celix_status_t status;

    status = wiringAdmin_stopWebserver(*admin);

    if (status == CELIX_SUCCESS) {
		celixThreadMutex_lock(&((*admin)->exportedWiringEndpointLock));
		hashMap_destroy((*admin)->wiringReceiveServices, false, false);
		hashMap_destroy((*admin)->wiringReceiveTracker, false, false);
		celixThreadMutex_unlock(&((*admin)->exportedWiringEndpointLock));
		celixThreadMutex_destroy(&((*admin)->exportedWiringEndpointLock));

		celixThreadMutex_lock(&((*admin)->importedWiringEndpointLock));
		hashMap_destroy((*admin)->wiringSendServices, false, false);
		hashMap_destroy((*admin)->wiringSendRegistrations, false, false);
		celixThreadMutex_unlock(&((*admin)->importedWiringEndpointLock));
		celixThreadMutex_destroy(&((*admin)->importedWiringEndpointLock));

		properties_destroy((*admin)->adminProperties);

		free(*admin);
		*admin = NULL;
    }

    return status;
}

celix_status_t pubsub_discovery_tmPublisherAnnounceAdded(void * handle, service_reference_pt reference, void * service) {
	celix_status_t status = CELIX_SUCCESS;

	pubsub_discovery_pt pubsub_discovery = (pubsub_discovery_pt)handle;
	publisher_endpoint_announce_pt listener = (publisher_endpoint_announce_pt)service;

	celixThreadMutex_lock(&pubsub_discovery->discoveredPubsMutex);
	celixThreadMutex_lock(&pubsub_discovery->listenerReferencesMutex);

	/* Notify the PSTM about discovered publisher endpoints */
	hash_map_iterator_pt iter = hashMapIterator_create(pubsub_discovery->discoveredPubs);
	while(hashMapIterator_hasNext(iter)){
		array_list_pt pubEP_list = (array_list_pt)hashMapIterator_nextValue(iter);
		int i;
		for(i=0;i<arrayList_size(pubEP_list);i++){
			pubsub_endpoint_pt pubEP = (pubsub_endpoint_pt)arrayList_get(pubEP_list,i);
			status += listener->announcePublisher(listener->handle, pubEP);
		}
	}

	hashMapIterator_destroy(iter);

	hashMap_put(pubsub_discovery->listenerReferences, reference, NULL);

	celixThreadMutex_unlock(&pubsub_discovery->listenerReferencesMutex);
	celixThreadMutex_unlock(&pubsub_discovery->discoveredPubsMutex);

	printf("PSD: pubsub_tm_announce_publisher added.\n");

	return status;
}

celix_status_t serviceRegistry_unregisterServices(service_registry_pt registry, bundle_pt bundle) {
	array_list_pt regs = NULL;
	unsigned int i;
	celixThreadMutex_lock(&registry->mutex);
	regs = (array_list_pt) hashMap_get(registry->serviceRegistrations, bundle);
	celixThreadMutex_unlock(&registry->mutex);
	
	for (i = 0; (regs != NULL) && i < arrayList_size(regs); i++) {
		service_registration_pt reg = arrayList_get(regs, i);
		if (serviceRegistration_isValid(reg)) {
			serviceRegistration_unregister(reg);
		}
	}

	if (regs != NULL && arrayList_isEmpty(regs)) {
	    celixThreadMutex_lock(&registry->mutex);
		array_list_pt removed = hashMap_remove(registry->serviceRegistrations, bundle);
		celixThreadMutex_unlock(&registry->mutex);
		arrayList_destroy(removed);
		removed = NULL;
	}

	celixThreadMutex_lock(&registry->mutex);
	hashMap_remove(registry->serviceRegistrations, bundle);
	celixThreadMutex_unlock(&registry->mutex);

	return CELIX_SUCCESS;
}

celix_status_t pubsubAdmin_removeSubscription(pubsub_admin_pt admin,pubsub_endpoint_pt subEP){
	celix_status_t status = CELIX_SUCCESS;

	printf("PSA_ZMQ: Removing subscription [FWUUID=%s bundleID=%ld topic=%s]\n",subEP->frameworkUUID,subEP->serviceID,subEP->topic);

	char* scope_topic = createScopeTopicKey(subEP->scope, subEP->topic);

	celixThreadMutex_lock(&admin->subscriptionsLock);
	topic_subscription_pt sub = (topic_subscription_pt)hashMap_get(admin->subscriptions,scope_topic);
	if(sub!=NULL){
		pubsub_topicDecreaseNrSubscribers(sub);
		if(pubsub_topicGetNrSubscribers(sub) == 0) {
			status = pubsub_topicSubscriptionRemoveSubscriber(sub,subEP);
		}
	}
	celixThreadMutex_unlock(&admin->subscriptionsLock);

	if(sub==NULL){
		/* Maybe the endpoint was pending */
		celixThreadMutex_lock(&admin->noSerializerPendingsLock);
		if(!arrayList_removeElement(admin->noSerializerSubscriptions, subEP)){
			status = CELIX_ILLEGAL_STATE;
		}
		celixThreadMutex_unlock(&admin->noSerializerPendingsLock);
	}

	free(scope_topic);



	return status;

}

celix_status_t serviceRegistry_getService(service_registry_pt registry, bundle_pt bundle, service_reference_pt reference, void **service) {
	celix_status_t status = CELIX_SUCCESS;
	service_registration_pt registration = NULL;
	*service = NULL;
	usage_count_pt usage = NULL;
	serviceReference_getServiceRegistration(reference, &registration);
	
	celixThreadMutex_lock(&registry->mutex);

	if (serviceRegistration_isValid(registration)) {
		status = serviceRegistry_getUsageCount(registry, bundle, reference, &usage);
		if (usage == NULL) {
			status = serviceRegistry_addUsageCount(registry, bundle, reference, &usage);
		}
		usage->count++;
		*service = usage->service;
	}
	celixThreadMutex_unlock(&registry->mutex);

	if ((usage != NULL) && (*service == NULL)) {
		status = serviceRegistration_getService(registration, bundle, service);
	}
	celixThreadMutex_lock(&registry->mutex);
	if ((!serviceRegistration_isValid(registration)) || (*service == NULL)) {
		serviceRegistry_flushUsageCount(registry, bundle, reference);
	} else {
		usage->service = *service;
	}
	celixThreadMutex_unlock(&registry->mutex);

	return status;
}

celix_status_t wiringTopologyManager_waRemoved(void * handle, service_reference_pt reference, void * service) {
    celix_status_t status = CELIX_SUCCESS;
    wiring_topology_manager_pt manager = handle;
    wiring_admin_service_pt wiringAdminService = (wiring_admin_service_pt) service;

    /* check whether one of the exported Wires can be exported here via the newly available wiringAdmin*/
    celixThreadMutex_lock(&manager->exportedWiringEndpointsLock);
    hash_map_iterator_pt iter = hashMapIterator_create(manager->exportedWiringEndpoints);
    while (hashMapIterator_hasNext(iter)) {
        hash_map_entry_pt entry = hashMapIterator_nextEntry(iter);
        hash_map_pt wiringAdminMap = hashMapEntry_getValue(entry);

        if (hashMap_containsKey(wiringAdminMap, wiringAdminService)) {
            wiring_endpoint_description_pt wEndpoint = (wiring_endpoint_description_pt) hashMap_remove(wiringAdminMap, wiringAdminService);

            status = wiringTopologyManager_notifyListenersWiringEndpointRemoved(manager, wEndpoint);

            if (status == CELIX_SUCCESS) {
                status = wiringAdminService->removeExportedWiringEndpoint(wiringAdminService->admin, wEndpoint);
            } else {
                printf("WTM: failed while removing WiringAdmin.\n");
            }
        }
    }

    hashMapIterator_destroy(iter);

    celixThreadMutex_unlock(&manager->exportedWiringEndpointsLock);

    /* Check if the added WA can match one of the imported WiringEndpoints */
    celixThreadMutex_lock(&manager->importedWiringEndpointsLock);
    iter = hashMapIterator_create(manager->importedWiringEndpoints);
    while (hashMapIterator_hasNext(iter)) {
        hash_map_entry_pt entry = hashMapIterator_nextEntry(iter);
        wiring_endpoint_description_pt importedWiringEndpointDesc = hashMapEntry_getKey(entry);
        array_list_pt wiringAdminList = hashMapEntry_getValue(entry);

        if (arrayList_contains(wiringAdminList, wiringAdminService)) {
            status = wiringAdminService->removeImportedWiringEndpoint(wiringAdminService->admin, importedWiringEndpointDesc);
            arrayList_removeElement(wiringAdminList, wiringAdminService);
        }

        if (status == CELIX_SUCCESS) {
            arrayList_add(wiringAdminList, wiringAdminService);
        }

    }
    hashMapIterator_destroy(iter);
    celixThreadMutex_unlock(&manager->importedWiringEndpointsLock);

    celixThreadMutex_lock(&manager->waListLock);
    arrayList_removeElement(manager->waList, wiringAdminService);
    celixThreadMutex_unlock(&manager->waListLock);

    printf("WTM: Removed WA\n");

    return status;
}

celix_status_t pubsub_discovery_stop(pubsub_discovery_pt ps_discovery) {
    celix_status_t status = CELIX_SUCCESS;

    const char* fwUUID = NULL;

    bundleContext_getProperty(ps_discovery->context, OSGI_FRAMEWORK_FRAMEWORK_UUID, &fwUUID);
    if (fwUUID == NULL) {
        printf("PSD: Cannot retrieve fwUUID.\n");
        return CELIX_INVALID_BUNDLE_CONTEXT;
    }

    celixThreadMutex_lock(&ps_discovery->watchersMutex);

    hash_map_iterator_pt iter = hashMapIterator_create(ps_discovery->watchers);
    while (hashMapIterator_hasNext(iter)) {
        struct watcher_info * wi = hashMapIterator_nextValue(iter);
        etcdWatcher_stop(wi->watcher);
    }
    hashMapIterator_destroy(iter);

    celixThreadMutex_lock(&ps_discovery->discoveredPubsMutex);

    /* Unexport all publishers for the local framework, and also delete from ETCD publisher belonging to the local framework */

    iter = hashMapIterator_create(ps_discovery->discoveredPubs);
    while (hashMapIterator_hasNext(iter)) {
        array_list_pt pubEP_list = (array_list_pt) hashMapIterator_nextValue(iter);

        int i;
        for (i = 0; i < arrayList_size(pubEP_list); i++) {
            pubsub_endpoint_pt pubEP = (pubsub_endpoint_pt) arrayList_get(pubEP_list, i);
            if (strcmp(pubEP->frameworkUUID, fwUUID) == 0) {
                etcdWriter_deletePublisherEndpoint(ps_discovery->writer, pubEP);
            } else {
                pubsub_discovery_informPublishersListeners(ps_discovery, pubEP, false);
                arrayList_remove(pubEP_list, i);
                pubsubEndpoint_destroy(pubEP);
                i--;
            }
        }
    }

    hashMapIterator_destroy(iter);

    celixThreadMutex_unlock(&ps_discovery->discoveredPubsMutex);
    etcdWriter_destroy(ps_discovery->writer);

    iter = hashMapIterator_create(ps_discovery->watchers);
    while (hashMapIterator_hasNext(iter)) {
        struct watcher_info * wi = hashMapIterator_nextValue(iter);
        etcdWatcher_destroy(wi->watcher);
    }
    hashMapIterator_destroy(iter);
    hashMap_destroy(ps_discovery->watchers, true, true);
    celixThreadMutex_unlock(&ps_discovery->watchersMutex);
    return status;
}

celix_status_t serviceRegistry_unregisterService(service_registry_pt registry, bundle_pt bundle, service_registration_pt registration) {
	// array_list_t clients;
	unsigned int i;
	array_list_pt regs;
	array_list_pt references = NULL;

	celixThreadMutex_lock(&registry->mutex);

	serviceRegistry_removeHook(registry, registration);

	regs = (array_list_pt) hashMap_get(registry->serviceRegistrations, bundle);
	if (regs != NULL) {
		arrayList_removeElement(regs, registration);
		hashMap_put(registry->serviceRegistrations, bundle, regs);
	}

	celixThreadMutex_unlock(&registry->mutex);

	if (registry->serviceChanged != NULL) {
		registry->serviceChanged(registry->framework, OSGI_FRAMEWORK_SERVICE_EVENT_UNREGISTERING, registration, NULL);
	}

	celixThreadMutex_lock(&registry->mutex);
	// unget service

	serviceRegistration_getServiceReferences(registration, &references);
	for (i = 0; i < arrayList_size(references); i++) {
		service_reference_pt reference = (service_reference_pt) arrayList_get(references, i);
		array_list_pt clients = NULL;
		unsigned int j;

		clients = serviceRegistry_getUsingBundles(registry, reference);
		for (j = 0; (clients != NULL) && (j < arrayList_size(clients)); j++) {
			bundle_pt client = (bundle_pt) arrayList_get(clients, j);
			bool ungetResult = true;
			while (ungetResult) {
				serviceRegistry_ungetService(registry, client, reference, &ungetResult);
			}
		}
		arrayList_destroy(clients);

		serviceReference_invalidate(reference);
	}
	arrayList_destroy(references);

	//TODO not needed, the registration is destroyed, any reference to the registration is invalid and will result in a segfault
	serviceRegistration_invalidate(registration);

	// serviceRegistration_destroy(registration);

	celixThreadMutex_unlock(&registry->mutex);

	return CELIX_SUCCESS;
}

celix_status_t wiringTopologyManager_wiringEndpointListenerAdded(void* handle, service_reference_pt reference, void* service) {
    celix_status_t status = CELIX_SUCCESS;
    wiring_topology_manager_pt manager = handle;
    char *scope = NULL;
    char* wtm = NULL;

    serviceReference_getProperty(reference, (char *) INAETICS_WIRING_ENDPOINT_LISTENER_SCOPE, &scope);
    serviceReference_getProperty(reference, "WTM", &wtm);

    if (wtm != NULL && strcmp(wtm, "true") == 0) {
        printf("WTM: Ignoring own ENDPOINT_LISTENER\n");
    }
    else {
        status = celixThreadMutex_lock(&manager->listenerListLock);

        if (status == CELIX_SUCCESS) {
            hashMap_put(manager->listenerList, reference, NULL);
            celixThreadMutex_unlock(&manager->listenerListLock);
        }

        filter_pt filter = filter_create(scope);
        status = celixThreadMutex_lock(&manager->exportedWiringEndpointsLock);

        if (status == CELIX_SUCCESS) {
            hash_map_iterator_pt propIter = hashMapIterator_create(manager->exportedWiringEndpoints);

            while (hashMapIterator_hasNext(propIter)) {
                hash_map_pt wiringAdminList = hashMapIterator_nextValue(propIter);
                hash_map_iterator_pt waIter = hashMapIterator_create(wiringAdminList);

                while (hashMapIterator_hasNext(waIter)) {
                    wiring_endpoint_description_pt wEndpoint = hashMapIterator_nextValue(waIter);

                    bool matchResult = false;
                    filter_match(filter, wEndpoint->properties, &matchResult);

                    if (matchResult) {
                        wiring_endpoint_listener_pt listener = (wiring_endpoint_listener_pt) service;
                        status = listener->wiringEndpointAdded(listener->handle, wEndpoint, scope);
                    }
                }
                hashMapIterator_destroy(waIter);
            }
            hashMapIterator_destroy(propIter);

            celixThreadMutex_unlock(&manager->exportedWiringEndpointsLock);
        }
        filter_destroy(filter);

    }


    return status;
}

celix_status_t serviceRegistry_registerServiceInternal(service_registry_pt registry, bundle_pt bundle, char * serviceName, void * serviceObject, properties_pt dictionary, bool isFactory, service_registration_pt *registration) {
	array_list_pt regs;
	celixThreadMutex_lock(&registry->mutex);

	if (isFactory) {
	    *registration = serviceRegistration_createServiceFactory(registry, bundle, serviceName, ++registry->currentServiceId, serviceObject, dictionary);
	} else {
	    *registration = serviceRegistration_create(registry, bundle, serviceName, ++registry->currentServiceId, serviceObject, dictionary);
	}

	serviceRegistry_addHooks(registry, serviceName, serviceObject, *registration);

	regs = (array_list_pt) hashMap_get(registry->serviceRegistrations, bundle);
	if (regs == NULL) {
		regs = NULL;
		arrayList_create(&regs);
	}
	arrayList_add(regs, *registration);
	hashMap_put(registry->serviceRegistrations, bundle, regs);

	celixThreadMutex_unlock(&registry->mutex);

	if (registry->serviceChanged != NULL) {
//		service_event_pt event = (service_event_pt) malloc(sizeof(*event));
//		event->type = REGISTERED;
//		event->reference = (*registration)->reference;
		registry->serviceChanged(registry->framework, OSGI_FRAMEWORK_SERVICE_EVENT_REGISTERED, *registration, NULL);
//		free(event);
//		event = NULL;
	}

	return CELIX_SUCCESS;
}

celix_status_t serviceRegistry_getRegisteredServices(service_registry_pt registry, bundle_pt bundle, array_list_pt *services) {
	celix_status_t status = CELIX_SUCCESS;

	celixThreadMutex_lock(&registry->mutex);
	array_list_pt regs = (array_list_pt) hashMap_get(registry->serviceRegistrations, bundle);
	if (regs != NULL) {
		unsigned int i;
		arrayList_create(services);
		
		for (i = 0; i < arrayList_size(regs); i++) {
			service_registration_pt reg = arrayList_get(regs, i);
			if (serviceRegistration_isValid(reg)) {
				service_reference_pt reference = NULL;
				status = serviceRegistry_createServiceReference(registry, bundle, reg, &reference);
				if (status == CELIX_SUCCESS) {
					arrayList_add(*services, reference);
				}
			}
		}
	}
	celixThreadMutex_unlock(&registry->mutex);

	framework_logIfError(logger, status, NULL, "Cannot get registered services");

	return status;
}

celix_status_t etcdWatcher_create(node_discovery_pt node_discovery, bundle_context_pt context, etcd_watcher_pt *watcher) {
    celix_status_t status = CELIX_SUCCESS;

    char* etcd_server = NULL;
    char* etcd_port_string = NULL;
    int etcd_port = 0;

    if (node_discovery == NULL) {
        return CELIX_BUNDLE_EXCEPTION;
    }

    (*watcher) = calloc(1, sizeof(struct etcd_watcher));

    if (*watcher) {
        (*watcher)->node_discovery = node_discovery;

        if ((bundleContext_getProperty(context, CFG_ETCD_SERVER_IP, &etcd_server) != CELIX_SUCCESS) || !etcd_server) {
            etcd_server = DEFAULT_ETCD_SERVER_IP;
        }

        if ((bundleContext_getProperty(context, CFG_ETCD_SERVER_PORT, &etcd_port_string) != CELIX_SUCCESS) || !etcd_port_string) {
            etcd_port = DEFAULT_ETCD_SERVER_PORT;
        } else {
            char* endptr = etcd_port_string;
            errno = 0;
            etcd_port = strtol(etcd_port_string, &endptr, 10);
            if (*endptr || errno != 0) {
                etcd_port = DEFAULT_ETCD_SERVER_PORT;
            }
        }

        if (etcd_init(etcd_server, etcd_port) == false) {
            status = CELIX_BUNDLE_EXCEPTION;
        } else {
            etcdWatcher_addOwnNode(*watcher);

            if ((status = celixThreadMutex_create(&(*watcher)->watcherLock, NULL)) != CELIX_SUCCESS) {
                return status;
            }

            if ((status = celixThreadMutex_lock(&(*watcher)->watcherLock)) != CELIX_SUCCESS) {
                return status;
            }

            if ((status = celixThread_create(&(*watcher)->watcherThread, NULL, etcdWatcher_run, *watcher)) != CELIX_SUCCESS) {
                return status;
            }

            (*watcher)->running = true;

            if ((status = celixThreadMutex_unlock(&(*watcher)->watcherLock)) != CELIX_SUCCESS) {
                return status;
            }
        }
    } else {
        status = CELIX_ENOMEM;
    }

    return status;
}

celix_status_t serviceRegistry_ungetServiceReference(service_registry_pt registry, bundle_pt owner, service_reference_pt reference) {
    celix_status_t status = CELIX_SUCCESS;

    bool valid = false;
    serviceRefernce_isValid(reference, &valid);
    if (valid) {
        bool ungetResult = true;
        while (ungetResult) {
            serviceRegistry_ungetService(registry, owner, reference, &ungetResult);
        }
    }

    celixThreadMutex_lock(&registry->referencesMapMutex);
    array_list_pt references = hashMap_get(registry->serviceReferences, owner);
    if (references != NULL) {
        arrayList_removeElement(references, reference);
        serviceReference_destroy(&reference);
        if (arrayList_size(references) > 0) {
            hashMap_put(registry->serviceReferences, owner, references);
        } else {
            array_list_pt removed = hashMap_remove(registry->serviceReferences, owner);
            arrayList_destroy(removed);
        }
    }
    celixThreadMutex_unlock(&registry->referencesMapMutex);

	return status;
}

celix_status_t etcdWriter_deletePublisherEndpoint(etcd_writer_pt writer, pubsub_endpoint_pt pubEP) {
	celix_status_t status = CELIX_SUCCESS;
	char *key = NULL;

	const char *rootPath = etcdWriter_getRootPath(writer->pubsub_discovery->context);

	asprintf(&key, "%s/%s/%s/%s/%ld", rootPath, pubEP->scope, pubEP->topic, pubEP->frameworkUUID, pubEP->serviceID);

	celixThreadMutex_lock(&writer->localPubsLock);
	for (unsigned int i = 0; i < arrayList_size(writer->localPubs); i++) {
		pubsub_endpoint_pt ep = arrayList_get(writer->localPubs, i);
		if (pubsubEndpoint_equals(ep, pubEP)) {
			arrayList_remove(writer->localPubs, i);
			pubsubEndpoint_destroy(ep);
			break;
		}
	}
	celixThreadMutex_unlock(&writer->localPubsLock);

	if (etcd_del(key)) {
		printf("Failed to remove key %s from ETCD\n",key);
		status = CELIX_ILLEGAL_ARGUMENT;
	}
	FREE_MEM(key);
	return status;
}

celix_status_t pubsubAdmin_closeAllSubscriptions(pubsub_admin_pt admin,char* scope,char* topic){
	celix_status_t status = CELIX_SUCCESS;

	printf("PSA_ZMQ: Closing all subscriptions\n");

	celixThreadMutex_lock(&admin->subscriptionsLock);
	char *scope_topic = createScopeTopicKey(scope, topic);
	hash_map_entry_pt sub_entry = (hash_map_entry_pt)hashMap_getEntry(admin->subscriptions,scope_topic);
	if(sub_entry!=NULL){
		char* topic = (char*)hashMapEntry_getKey(sub_entry);

		topic_subscription_pt ts = (topic_subscription_pt)hashMapEntry_getValue(sub_entry);
		status += pubsub_topicSubscriptionStop(ts);
		disconnectTopicPubSubFromSerializer(admin, ts, false);
		status += pubsub_topicSubscriptionDestroy(ts);
		hashMap_remove(admin->subscriptions,scope_topic);
		free(topic);

	}
	free(scope_topic);
	celixThreadMutex_unlock(&admin->subscriptionsLock);

	return status;

}

celix_status_t pubsub_topicSubscriptionDisconnectPublisher(topic_subscription_pt ts, char* pubURL){
	printf("pubsub_topicSubscriptionDisconnectPublisher : pubURL = %s\n", pubURL);
	celix_status_t status = CELIX_SUCCESS;
	struct epoll_event ev;
	memset(&ev, 0, sizeof(ev));

	celixThreadMutex_lock(&ts->socketMap_lock);

	if (hashMap_containsKey(ts->socketMap, pubURL)){

#if defined(__APPLE__) && defined(__MACH__)
		//TODO: Use kqueue for OSX
#else
		int *s = hashMap_remove(ts->socketMap, pubURL);
		if(epoll_ctl(ts->topicEpollFd, EPOLL_CTL_DEL, *s, &ev) == -1) {
			printf("in if error()\n");
			perror("epoll_ctl() EPOLL_CTL_DEL");
			status = CELIX_SERVICE_EXCEPTION;
		}
		free(s);
#endif

	}

	celixThreadMutex_unlock(&ts->socketMap_lock);

	return status;
}

/**
 * Destroys and frees up memory for a given endpoint_discovery_poller struct.
 */
celix_status_t endpointDiscoveryPoller_destroy(endpoint_discovery_poller_pt poller) {
    celix_status_t status;

    poller->running = false;

    celixThread_join(poller->pollerThread, NULL);

    hash_map_iterator_pt iterator = hashMapIterator_create(poller->entries);
	while (hashMapIterator_hasNext(iterator)) {
		hash_map_entry_pt entry = hashMapIterator_nextEntry(iterator);

		if ( endpointDiscoveryPoller_removeDiscoveryEndpoint(poller, (char*) hashMapEntry_getKey(entry)) == CELIX_SUCCESS) {
			hashMapIterator_destroy(iterator);
			iterator = hashMapIterator_create(poller->entries);
		}
	}
	hashMapIterator_destroy(iterator);

	status = celixThreadMutex_lock(&poller->pollerLock);

	if (status != CELIX_SUCCESS) {
		return CELIX_BUNDLE_EXCEPTION;
	}

	hashMap_destroy(poller->entries, true, false);

    status = celixThreadMutex_unlock(&poller->pollerLock);

    poller->loghelper = NULL;

    free(poller);

	return status;
}

void serviceRegistry_ungetServices(service_registry_pt registry, bundle_pt bundle) {
	array_list_pt fusages;
	array_list_pt usages;
	unsigned int i;

	celixThreadMutex_lock(&registry->mutex);
	usages = hashMap_get(registry->inUseMap, bundle);
	celixThreadMutex_unlock(&registry->mutex);

	if (usages == NULL || arrayList_isEmpty(usages)) {
		return;
	}

	// usage arrays?
	fusages = arrayList_clone(usages);
	
	for (i = 0; i < arrayList_size(fusages); i++) {
		usage_count_pt usage = arrayList_get(fusages, i);
		service_reference_pt reference = usage->reference;
		bool ungetResult = true;
		while (ungetResult) {
			serviceRegistry_ungetService(registry, bundle, reference, &ungetResult);
		}
	}

	arrayList_destroy(fusages);
}

celix_status_t bundle_unlock(bundle_pt bundle, bool *unlocked) {
	celix_status_t status = CELIX_SUCCESS;

	bool equals;

	celixThreadMutex_lock(&bundle->lock);

	if (bundle->lockCount == 0) {
		*unlocked = false;
	} else {
		status = thread_equalsSelf(bundle->lockThread, &equals);
		if (status == CELIX_SUCCESS) {
			if ((bundle->lockCount > 0) && !equals) {
				return false;
			}
			bundle->lockCount--;
			if (bundle->lockCount == 0) {
				bundle->lockThread = 0;
			}
			*unlocked = true;
		}
	}

	celixThreadMutex_unlock(&bundle->lock);

	framework_logIfError(bundle->framework->logger, status, NULL, "Failed to unlock bundle");

	return status;
}