| /* |
| * Copyright 2004,2005 The Apache Software Foundation. |
| * |
| * Licensed under the Apache License, Version 2.0 (the "License"); |
| * you may not use this file except in compliance with the License. |
| * You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| |
| #include <savan_subs_mgr.h> |
| #include <axutil_log.h> |
| #include <axutil_hash.h> |
| #include <axutil_property.h> |
| #include <axutil_types.h> |
| #include <axutil_file_handler.h> |
| #include <platforms/axutil_platform_auto_sense.h> |
| #include <savan_constants.h> |
| #include <savan_util.h> |
| #include <savan_error.h> |
| #include <remote_registry.h> |
| #include <remote_registry_resource.h> |
| |
| #define SYNAPSE_NAMESPACE "http://ws.apache.org/ns/synapse" |
| #define SYNAPSE_NS_PREFIX "syn" |
| #define ELEM_NAME_SUBSCRIPTION "subscription" |
| #define ELEM_NAME_ENDPOINT "endpoint" |
| #define ELEM_NAME_ADDRESS "address" |
| #define ATTR_NAME_URL "uri" |
| #define EPR_TYPE "application/vnd.epr" |
| #define TOPIC_INDEX_PARENT_PATH "/eventing/index" |
| #define SUBSCRIPTION_COLLECTION_NAME "system.subscriptions" |
| #define TOPIC_INDEX "/eventing/index/TopicIndex" |
| |
| /** |
| * Savan registry based subscription manager dependes on the WSO2 registry for subscription storation. |
| * This use WSF/C registry cache client to communicate with the WSO2 registry. Registry cache |
| * client reduce the overhead of each time fetching records from the registry which is very |
| * expensive by caching the records locally. |
| */ |
| /** |
| * @brief Savan Registry Storage Manager Struct Impl |
| * Savan Registry Storage Manager |
| */ |
| typedef struct savan_registry_subs_mgr |
| { |
| savan_subs_mgr_t subsmgr; |
| axis2_char_t *reg_url; |
| axis2_char_t *username; |
| axis2_char_t *password; |
| axis2_conf_t *conf; |
| remote_registry_t *remote_registry; |
| } savan_registry_subs_mgr_t; |
| |
| typedef AXIS2_DECLARE_DATA struct savan_registry_subs_mgr_args |
| { |
| const axutil_env_t *env; |
| void *data; |
| } savan_registry_subs_mgr_args_t; |
| |
| #define SAVAN_INTF_TO_IMPL(subsmgr) ((savan_registry_subs_mgr_t *) subsmgr) |
| |
| static savan_subscriber_t *savan_registry_subs_mgr_extract_subscriber( |
| const axutil_env_t *env, |
| remote_registry_resource_t *resource); |
| |
| static axis2_char_t *savan_registry_subs_mgr_serialize_endpoint( |
| const axutil_env_t *env, |
| const savan_subscriber_t *subscriber); |
| |
| static axis2_status_t |
| savan_registry_subs_mgr_init_resource( |
| savan_subs_mgr_t *subsmgr, |
| const axutil_env_t *env); |
| |
| AXIS2_EXTERN void AXIS2_CALL |
| savan_registry_subs_mgr_free( |
| savan_subs_mgr_t *subsmgr, |
| const axutil_env_t *env); |
| |
| AXIS2_EXTERN axis2_status_t AXIS2_CALL |
| savan_registry_subs_mgr_insert_subscriber( |
| savan_subs_mgr_t *subsmgr, |
| const axutil_env_t *env, |
| savan_subscriber_t *subscriber); |
| |
| AXIS2_EXTERN axis2_status_t AXIS2_CALL |
| savan_registry_subs_mgr_update_subscriber( |
| savan_subs_mgr_t *subsmgr, |
| const axutil_env_t *env, |
| savan_subscriber_t *subscriber); |
| |
| AXIS2_EXTERN axis2_status_t AXIS2_CALL |
| savan_registry_subs_mgr_remove_subscriber( |
| savan_subs_mgr_t *subsmgr, |
| const axutil_env_t *env, |
| const axis2_char_t *subscriber_id); |
| |
| AXIS2_EXTERN savan_subscriber_t *AXIS2_CALL |
| savan_registry_subs_mgr_retrieve_subscriber( |
| savan_subs_mgr_t *subsmgr, |
| const axutil_env_t *env, |
| const axis2_char_t *subscriber_id); |
| |
| AXIS2_EXTERN axutil_array_list_t * AXIS2_CALL |
| savan_registry_subs_mgr_retrieve_all_subscribers( |
| savan_subs_mgr_t *subsmgr, |
| const axutil_env_t *env, |
| const axis2_char_t *filter); |
| |
| static const savan_subs_mgr_ops_t subs_mgr_ops = |
| { |
| savan_registry_subs_mgr_free, |
| savan_registry_subs_mgr_insert_subscriber, |
| savan_registry_subs_mgr_update_subscriber, |
| savan_registry_subs_mgr_remove_subscriber, |
| savan_registry_subs_mgr_retrieve_subscriber, |
| savan_registry_subs_mgr_retrieve_all_subscribers, |
| NULL |
| }; |
| |
| AXIS2_EXTERN savan_subs_mgr_t * AXIS2_CALL |
| savan_subs_mgr_create( |
| const axutil_env_t *env, |
| axis2_conf_t *conf) |
| { |
| savan_registry_subs_mgr_t *subsmgrimpl = NULL; |
| axis2_status_t status = AXIS2_FAILURE; |
| |
| subsmgrimpl = AXIS2_MALLOC(env->allocator, sizeof(savan_registry_subs_mgr_t)); |
| if (!subsmgrimpl) |
| { |
| AXIS2_HANDLE_ERROR(env, SAVAN_ERROR_STORAGE_MANAGER_CREATION_FAILED, AXIS2_FAILURE); |
| return NULL; |
| } |
| |
| memset ((void *) subsmgrimpl, 0, sizeof(savan_registry_subs_mgr_t)); |
| |
| subsmgrimpl->remote_registry = NULL; |
| subsmgrimpl->reg_url = savan_util_get_resource_connection_string(env, conf); |
| subsmgrimpl->username = axutil_strdup(env, savan_util_get_resource_username(env, conf)); |
| subsmgrimpl->password = axutil_strdup(env, savan_util_get_resource_password(env, conf)); |
| subsmgrimpl->conf = conf; |
| subsmgrimpl->subsmgr.ops = &subs_mgr_ops; |
| |
| status = savan_registry_subs_mgr_init_resource((savan_subs_mgr_t *) subsmgrimpl, env); |
| if(status != AXIS2_SUCCESS) |
| { |
| savan_registry_subs_mgr_free((savan_subs_mgr_t *) subsmgrimpl, env); |
| return NULL; |
| } |
| return (savan_subs_mgr_t *) subsmgrimpl; |
| } |
| |
| AXIS2_EXTERN savan_subs_mgr_t * AXIS2_CALL |
| savan_subs_mgr_create_with_connection_info( |
| const axutil_env_t *env, |
| axis2_char_t *connection_string, |
| axis2_char_t *username, |
| axis2_char_t *password) |
| { |
| savan_registry_subs_mgr_t *subsmgrimpl = NULL; |
| axis2_status_t status = AXIS2_FAILURE; |
| |
| subsmgrimpl = AXIS2_MALLOC(env->allocator, sizeof(savan_registry_subs_mgr_t)); |
| if (!subsmgrimpl) |
| { |
| AXIS2_HANDLE_ERROR(env, SAVAN_ERROR_STORAGE_MANAGER_CREATION_FAILED, AXIS2_FAILURE); |
| return NULL; |
| } |
| |
| memset ((void *) subsmgrimpl, 0, sizeof(savan_registry_subs_mgr_t)); |
| |
| subsmgrimpl->remote_registry = NULL; |
| subsmgrimpl->reg_url = axutil_strdup(env, connection_string); |
| subsmgrimpl->username = axutil_strdup(env, username); |
| subsmgrimpl->password = axutil_strdup(env, password); |
| subsmgrimpl->conf = NULL; |
| subsmgrimpl->subsmgr.ops = &subs_mgr_ops; |
| |
| status = savan_registry_subs_mgr_init_resource((savan_subs_mgr_t *) subsmgrimpl, env); |
| if(status != AXIS2_SUCCESS) |
| { |
| savan_registry_subs_mgr_free((savan_subs_mgr_t *) subsmgrimpl, env); |
| return NULL; |
| } |
| return (savan_subs_mgr_t *) subsmgrimpl; |
| } |
| |
| AXIS2_EXTERN void AXIS2_CALL |
| savan_registry_subs_mgr_free( |
| savan_subs_mgr_t *subsmgr, |
| const axutil_env_t *env) |
| { |
| savan_registry_subs_mgr_t *subsmgrimpl = NULL; |
| subsmgrimpl = SAVAN_INTF_TO_IMPL(subsmgr); |
| |
| AXIS2_LOG_TRACE(env->log, AXIS2_LOG_SI, "[savan] Entry:savan_registry_subs_mgr_free"); |
| |
| if(subsmgrimpl->reg_url) |
| { |
| AXIS2_FREE(env->allocator, subsmgrimpl->reg_url); |
| subsmgrimpl->reg_url = NULL; |
| } |
| |
| if(subsmgrimpl->username) |
| { |
| AXIS2_FREE(env->allocator, subsmgrimpl->username); |
| subsmgrimpl->username = NULL; |
| } |
| |
| if(subsmgrimpl->password) |
| { |
| AXIS2_FREE(env->allocator, subsmgrimpl->password); |
| subsmgrimpl->password = NULL; |
| } |
| |
| subsmgrimpl->conf = NULL; |
| |
| if(subsmgrimpl) |
| { |
| AXIS2_FREE(env->allocator, subsmgrimpl); |
| subsmgrimpl = NULL; |
| } |
| |
| AXIS2_LOG_TRACE(env->log, AXIS2_LOG_SI, "[savan] Exit:savan_registry_subs_mgr_free"); |
| } |
| |
| AXIS2_EXTERN axis2_status_t AXIS2_CALL |
| savan_registry_subs_mgr_insert_subscriber( |
| savan_subs_mgr_t *subsmgr, |
| const axutil_env_t *env, |
| savan_subscriber_t *subscriber) |
| { |
| axis2_char_t *subscription_id = NULL; |
| axis2_char_t *expires = NULL; |
| axis2_char_t *filter = NULL; |
| axis2_char_t *id = NULL; |
| axis2_char_t *path = NULL; |
| remote_registry_resource_t *res = NULL; |
| axutil_hash_t *properties = NULL; |
| char *content = NULL; |
| |
| savan_registry_subs_mgr_t *subsmgrimpl = NULL; |
| subsmgrimpl = SAVAN_INTF_TO_IMPL(subsmgr); |
| |
| AXIS2_LOG_TRACE(env->log, AXIS2_LOG_SI, |
| "[savan] Entry:savan_registry_subs_mgr_insert_subscriber"); |
| |
| subscription_id = savan_subscriber_get_id(subscriber, env); |
| filter = savan_subscriber_get_filter(subscriber, env); |
| path = axutil_strcat(env, AXIS2_PATH_SEP_STR, filter, AXIS2_PATH_SEP_STR, |
| SUBSCRIPTION_COLLECTION_NAME, AXIS2_PATH_SEP_STR, subscription_id, NULL); |
| id = axutil_strcat(env, subsmgrimpl->reg_url, AXIS2_PATH_SEP_STR, filter, |
| AXIS2_PATH_SEP_STR, SUBSCRIPTION_COLLECTION_NAME, AXIS2_PATH_SEP_STR, subscription_id, |
| NULL); |
| |
| res = remote_registry_resource_create(env); |
| content = savan_registry_subs_mgr_serialize_endpoint(env, subscriber); |
| remote_registry_resource_set_content(res, env, content); |
| remote_registry_resource_set_content_len(res, env, axutil_strlen(content)); |
| remote_registry_resource_set_media_type(res, env, EPR_TYPE); |
| remote_registry_resource_set_description(res, env, ""); |
| |
| properties = axutil_hash_make(env); |
| if(properties) |
| { |
| axis2_char_t *endto = NULL; |
| axis2_endpoint_ref_t *endto_epr = NULL; |
| axis2_char_t *filter_dialect = NULL; |
| |
| |
| expires = savan_subscriber_get_expires(subscriber, env); |
| if(expires) |
| { |
| axutil_hash_set(properties, axutil_strdup(env, "expires"), AXIS2_HASH_KEY_STRING, |
| axutil_strdup(env, expires)); |
| } |
| else |
| { |
| axutil_hash_set(properties, axutil_strdup(env, "expires"), AXIS2_HASH_KEY_STRING, |
| axutil_strdup(env, "*")); |
| } |
| |
| axutil_hash_set(properties, axutil_strdup(env, "staticFlag"), AXIS2_HASH_KEY_STRING, |
| axutil_strdup(env, "false")); |
| axutil_hash_set(properties, axutil_strdup(env, "filterValue"), AXIS2_HASH_KEY_STRING, |
| axutil_strdup(env, filter)); |
| |
| endto_epr = savan_subscriber_get_end_to(subscriber, env); |
| if(endto_epr) |
| { |
| endto = (axis2_char_t *) axis2_endpoint_ref_get_address(endto_epr, env); |
| } |
| if(endto) |
| { |
| axutil_hash_set(properties, axutil_strdup(env, "subManagerURI"), AXIS2_HASH_KEY_STRING, |
| axutil_strdup(env, endto)); |
| } |
| |
| filter_dialect = savan_subscriber_get_filter_dialect(subscriber, env); |
| if(filter_dialect) |
| { |
| axutil_hash_set(properties, axutil_strdup(env, "filterDialect"), AXIS2_HASH_KEY_STRING, |
| axutil_strdup(env, filter_dialect)); |
| } |
| |
| remote_registry_resource_set_properties(res, env, properties); |
| } |
| |
| remote_registry_put(subsmgrimpl->remote_registry, env, path, res); |
| if(id) |
| { |
| AXIS2_FREE(env->allocator, id); |
| } |
| if(path) |
| { |
| AXIS2_FREE(env->allocator, path); |
| path = NULL; |
| } |
| if(res) |
| { |
| remote_registry_resource_free(res, env); |
| res = NULL; |
| } |
| |
| res = remote_registry_get(subsmgrimpl->remote_registry, env, TOPIC_INDEX, NULL); |
| if(!res) |
| { |
| AXIS2_HANDLE_ERROR(env, SAVAN_ERROR_SUBSCRIBER_INSERT_ERROR, AXIS2_FAILURE); |
| AXIS2_LOG_ERROR(env->log, AXIS2_LOG_SI, "[savan] Could not retrive resource TopicIndex"); |
| return AXIS2_FAILURE; |
| } |
| id = axutil_strcat(env, subsmgrimpl->reg_url, TOPIC_INDEX, NULL); |
| properties = remote_registry_resource_get_properties(res, env); |
| if(properties) |
| { |
| path = axutil_strcat(env, AXIS2_PATH_SEP_STR, filter, AXIS2_PATH_SEP_STR, |
| SUBSCRIPTION_COLLECTION_NAME, NULL); |
| axutil_hash_set(properties, subscription_id, AXIS2_HASH_KEY_STRING, path); |
| remote_registry_resource_set_properties(res, env, properties); |
| } |
| |
| remote_registry_resource_set_content(res, env, NULL); |
| remote_registry_resource_set_content_len(res, env, 0); |
| remote_registry_put(subsmgrimpl->remote_registry, env, TOPIC_INDEX, res); |
| |
| if(id) |
| { |
| AXIS2_FREE(env->allocator, id); |
| } |
| if(res) |
| { |
| remote_registry_resource_free(res, env); |
| res = NULL; |
| } |
| |
| AXIS2_LOG_TRACE(env->log, AXIS2_LOG_SI, |
| "[savan] Exit:savan_registry_subs_mgr_insert_subscriber"); |
| return AXIS2_SUCCESS; |
| } |
| |
| AXIS2_EXTERN axis2_status_t AXIS2_CALL |
| savan_registry_subs_mgr_update_subscriber( |
| savan_subs_mgr_t *subsmgr, |
| const axutil_env_t *env, |
| savan_subscriber_t *subscriber) |
| { |
| axis2_char_t *expires = NULL; |
| axis2_char_t *filter = NULL; |
| axis2_char_t *subscriber_id = NULL; |
| axis2_char_t *path = NULL; |
| remote_registry_resource_t *res = NULL; |
| remote_registry_resource_t *index_res = NULL; |
| axutil_hash_t *properties = NULL; |
| axis2_char_t *val = NULL; |
| |
| savan_registry_subs_mgr_t *subsmgrimpl = NULL; |
| subsmgrimpl = SAVAN_INTF_TO_IMPL(subsmgr); |
| |
| AXIS2_LOG_TRACE(env->log, AXIS2_LOG_SI, |
| "[savan] Entry:savan_registry_subs_mgr_update_subscriber"); |
| |
| index_res = remote_registry_get(subsmgrimpl->remote_registry, env, TOPIC_INDEX, NULL); |
| if(!index_res) |
| { |
| AXIS2_HANDLE_ERROR(env, SAVAN_ERROR_SUBSCRIBER_INSERT_ERROR, AXIS2_FAILURE); |
| AXIS2_LOG_ERROR(env->log, AXIS2_LOG_SI, "[savan] Could not retrive resource TopicIndex"); |
| return AXIS2_FAILURE; |
| } |
| |
| subscriber_id = savan_subscriber_get_id(subscriber, env); |
| filter = savan_subscriber_get_filter(subscriber, env); |
| val = remote_registry_resource_get_property(index_res, env, subscriber_id); |
| if(val) |
| { |
| path = axutil_strcat(env, val, AXIS2_PATH_SEP_STR, subscriber_id, NULL); |
| } |
| |
| res = remote_registry_get(subsmgrimpl->remote_registry, env, path, NULL); |
| if(!res) |
| { |
| AXIS2_HANDLE_ERROR(env, SAVAN_ERROR_SUBSCRIBER_UPDATE_ERROR, AXIS2_FAILURE); |
| AXIS2_LOG_ERROR(env->log, AXIS2_LOG_SI, |
| "[savan] Could not retrive resource subscriber path:%s in registry", path); |
| return AXIS2_FAILURE; |
| } |
| |
| remote_registry_resource_set_description(res, env, ""); |
| properties = remote_registry_resource_get_properties(res, env); |
| if(properties) |
| { |
| axis2_char_t *endto = NULL; |
| axis2_endpoint_ref_t *endto_epr = NULL; |
| axis2_char_t *filter_dialect = NULL; |
| |
| expires = savan_subscriber_get_expires(subscriber, env); |
| if(expires) |
| { |
| axutil_hash_set(properties, axutil_strdup(env, "expires"), AXIS2_HASH_KEY_STRING, |
| axutil_strdup(env, expires)); |
| } |
| |
| axutil_hash_set(properties, axutil_strdup(env, "staticFlag"), AXIS2_HASH_KEY_STRING, |
| axutil_strdup(env, "false")); |
| axutil_hash_set(properties, axutil_strdup(env, "filterValue"), AXIS2_HASH_KEY_STRING, |
| axutil_strdup(env, filter)); |
| |
| endto_epr = savan_subscriber_get_end_to(subscriber, env); |
| if(endto_epr) |
| { |
| endto = (axis2_char_t *) axis2_endpoint_ref_get_address(endto_epr, env); |
| } |
| if(endto) |
| { |
| axutil_hash_set(properties, axutil_strdup(env, "subManagerURI"), AXIS2_HASH_KEY_STRING, |
| axutil_strdup(env, endto)); |
| } |
| |
| filter_dialect = savan_subscriber_get_filter_dialect(subscriber, env); |
| if(filter_dialect) |
| { |
| axutil_hash_set(properties, axutil_strdup(env, "filterDialect"), AXIS2_HASH_KEY_STRING, |
| axutil_strdup(env, filter_dialect)); |
| } |
| |
| remote_registry_resource_set_properties(res, env, properties); |
| } |
| |
| remote_registry_put(subsmgrimpl->remote_registry, env, path, res); |
| |
| if(path) |
| { |
| AXIS2_FREE(env->allocator, path); |
| path = NULL; |
| } |
| if(res) |
| { |
| remote_registry_resource_free(res, env); |
| res = NULL; |
| } |
| |
| savan_subscriber_set_renew_status(subscriber, env, AXIS2_TRUE); |
| AXIS2_LOG_TRACE(env->log, AXIS2_LOG_SI, |
| "[savan] Exit:savan_registry_subs_mgr_update_subscriber"); |
| return AXIS2_SUCCESS; |
| } |
| |
| AXIS2_EXTERN axis2_status_t AXIS2_CALL |
| savan_registry_subs_mgr_remove_subscriber( |
| savan_subs_mgr_t *subsmgr, |
| const axutil_env_t *env, |
| const axis2_char_t *subscriber_id) |
| { |
| remote_registry_resource_t *res = NULL; |
| axis2_char_t *val; |
| axis2_status_t status = AXIS2_FAILURE; |
| axis2_char_t *path = NULL; |
| |
| savan_registry_subs_mgr_t *subsmgrimpl = NULL; |
| subsmgrimpl = SAVAN_INTF_TO_IMPL(subsmgr); |
| |
| AXIS2_LOG_TRACE(env->log, AXIS2_LOG_SI, |
| "[savan] Entry:savan_registry_subs_mgr_remove_subscriber"); |
| |
| res = remote_registry_get(subsmgrimpl->remote_registry, env, TOPIC_INDEX, NULL); |
| if(!res) |
| { |
| AXIS2_HANDLE_ERROR(env, SAVAN_ERROR_SUBSCRIBER_UPDATE_ERROR, AXIS2_FAILURE); |
| AXIS2_LOG_ERROR(env->log, AXIS2_LOG_SI, |
| "[savan] Could not retrive resource TopicIndex"); |
| return AXIS2_FAILURE; |
| } |
| |
| val = remote_registry_resource_get_property(res, env, subscriber_id); |
| if(val) |
| { |
| path = axutil_strcat(env, val, AXIS2_PATH_SEP_STR, subscriber_id, NULL); |
| status = remote_registry_delete(subsmgrimpl->remote_registry, env, path); |
| } |
| |
| remote_registry_resource_free(res, env); |
| if(AXIS2_SUCCESS != status) |
| { |
| AXIS2_HANDLE_ERROR(env, SAVAN_ERROR_SUBSCRIBER_REMOVE_ERROR, AXIS2_FAILURE); |
| AXIS2_LOG_ERROR(env->log, AXIS2_LOG_SI, |
| "[savan] Could not remove subscriber for id %s", subscriber_id); |
| return AXIS2_FAILURE; |
| } |
| |
| res = remote_registry_get(subsmgrimpl->remote_registry, env, TOPIC_INDEX, NULL); |
| if(!res) |
| { |
| AXIS2_HANDLE_ERROR(env, SAVAN_ERROR_SUBSCRIBER_UPDATE_ERROR, AXIS2_FAILURE); |
| AXIS2_LOG_ERROR(env->log, AXIS2_LOG_SI, |
| "[savan] Could not retrive resource TopicIndex"); |
| return AXIS2_FAILURE; |
| } |
| |
| remote_registry_resource_remove_property(res, env, subscriber_id); |
| remote_registry_put(subsmgrimpl->remote_registry, env, TOPIC_INDEX, res); |
| remote_registry_resource_free(res, env); |
| |
| AXIS2_LOG_TRACE(env->log, AXIS2_LOG_SI, |
| "[savan] Exit:savan_registry_subs_mgr_remove_subscriber"); |
| return AXIS2_SUCCESS; |
| } |
| |
| AXIS2_EXTERN savan_subscriber_t *AXIS2_CALL |
| savan_registry_subs_mgr_retrieve_subscriber( |
| savan_subs_mgr_t *subsmgr, |
| const axutil_env_t *env, |
| const axis2_char_t *subscriber_id) |
| { |
| remote_registry_resource_t *res = NULL; |
| remote_registry_resource_t *root_res = NULL; |
| axis2_char_t *val; |
| axis2_char_t *path = NULL; |
| savan_subscriber_t *subscriber = NULL; |
| |
| savan_registry_subs_mgr_t *subsmgrimpl = NULL; |
| subsmgrimpl = SAVAN_INTF_TO_IMPL(subsmgr); |
| |
| |
| AXIS2_LOG_TRACE(env->log, AXIS2_LOG_SI, |
| "[savan] Entry:savan_registry_subs_mgr_retrieve_subscriber"); |
| |
| root_res = remote_registry_get(subsmgrimpl->remote_registry, env, TOPIC_INDEX, NULL); |
| if(!root_res) |
| { |
| AXIS2_HANDLE_ERROR(env, SAVAN_ERROR_SUBSCRIBER_UPDATE_ERROR, AXIS2_FAILURE); |
| AXIS2_LOG_ERROR(env->log, AXIS2_LOG_SI, |
| "[savan] Could not retrive resource TopicIndex"); |
| return AXIS2_FAILURE; |
| } |
| |
| val = remote_registry_resource_get_property(root_res, env, subscriber_id); |
| if(val) |
| { |
| path = axutil_strcat(env, val, AXIS2_PATH_SEP_STR, subscriber_id, NULL); |
| res = remote_registry_get(subsmgrimpl->remote_registry, env, path, NULL); |
| } |
| |
| if(res) |
| { |
| subscriber = savan_registry_subs_mgr_extract_subscriber(env, res); |
| if(subscriber) |
| { |
| savan_subscriber_set_id(subscriber, env, subscriber_id); |
| } |
| } |
| |
| AXIS2_LOG_TRACE(env->log, AXIS2_LOG_SI, |
| "[savan] Exit:savan_registry_subs_mgr_retrieve_subscriber"); |
| return subscriber; |
| } |
| |
| AXIS2_EXTERN axutil_array_list_t * AXIS2_CALL |
| savan_registry_subs_mgr_retrieve_all_subscribers( |
| savan_subs_mgr_t *subsmgr, |
| const axutil_env_t *env, |
| const axis2_char_t *filter) |
| { |
| remote_registry_resource_t *root_res = NULL; |
| axis2_char_t *path = NULL; |
| axutil_array_list_t *data_list = NULL; |
| |
| savan_registry_subs_mgr_t *subsmgrimpl = NULL; |
| subsmgrimpl = SAVAN_INTF_TO_IMPL(subsmgr); |
| |
| AXIS2_LOG_TRACE(env->log, AXIS2_LOG_SI, |
| "[savan] Entry:savan_registry_subs_mgr_retrieve_all_subscribers"); |
| |
| /* Get subscribers for the filter from registry. Eg. /weather/4/system.subscriptions */ |
| if(filter) |
| { |
| root_res = remote_registry_get(subsmgrimpl->remote_registry, env, |
| (axis2_char_t *) filter, NULL); |
| if(root_res) |
| { |
| axutil_array_list_t *child_entries = NULL; |
| int i = 0; |
| |
| child_entries = remote_registry_resource_get_entries(root_res, env); |
| if(child_entries) |
| { |
| data_list = axutil_array_list_create(env, 0); |
| |
| /* load the child entries recursively */ |
| for(i = 0; i < axutil_array_list_size(child_entries, env); i ++) |
| { |
| remote_registry_resource_t *res = NULL; |
| |
| res = (remote_registry_resource_t*)axutil_array_list_get(child_entries, env, i); |
| if(res) |
| { |
| savan_subscriber_t *subscriber = NULL; |
| |
| subscriber = savan_registry_subs_mgr_extract_subscriber(env, res); |
| axutil_array_list_add(data_list, env, subscriber); |
| } |
| } |
| } |
| } |
| } |
| else /* Get subscribers for the topic root from registry(/eventing/index/TopicIndex);*/ |
| { |
| axutil_hash_t *properties = NULL; |
| |
| root_res = remote_registry_get(subsmgrimpl->remote_registry, env, TOPIC_INDEX, NULL); |
| if(!root_res) |
| { |
| AXIS2_HANDLE_ERROR(env, SAVAN_ERROR_SUBSCRIBER_UPDATE_ERROR, AXIS2_FAILURE); |
| AXIS2_LOG_ERROR(env->log, AXIS2_LOG_SI, |
| "[savan] Could not retrive resource TopicIndex"); |
| return AXIS2_FAILURE; |
| } |
| properties = remote_registry_resource_get_properties(root_res, env); |
| if(properties) |
| { |
| savan_subscriber_t *subscriber = NULL; |
| axutil_hash_index_t *hi; |
| void *val; |
| void *key; |
| |
| data_list = axutil_array_list_create(env, 0); |
| |
| for (hi = axutil_hash_first(properties, env); hi; hi = axutil_hash_next(env, hi)) |
| { |
| remote_registry_resource_t *res = NULL; |
| axis2_char_t *subscriber_id = NULL; |
| |
| axutil_hash_this(hi, (const void**)&key, NULL, &val); |
| subscriber_id = (axis2_char_t *) key; |
| path = axutil_strcat(env, val, AXIS2_PATH_SEP_STR, key, NULL); |
| res = remote_registry_get(subsmgrimpl->remote_registry, env, path, NULL); |
| if(res) |
| { |
| if(res) |
| { |
| subscriber = savan_registry_subs_mgr_extract_subscriber(env, res); |
| savan_subscriber_set_id(subscriber, env, subscriber_id); |
| axutil_array_list_add(data_list, env, subscriber); |
| } |
| } |
| } |
| } |
| } |
| |
| AXIS2_LOG_TRACE(env->log, AXIS2_LOG_SI, |
| "[savan] Exit:savan_registry_subs_mgr_retrieve_all_subscribers"); |
| return data_list; |
| } |
| |
| static axis2_status_t |
| savan_registry_subs_mgr_init_resource( |
| savan_subs_mgr_t *subsmgr, |
| const axutil_env_t *env) |
| { |
| remote_registry_resource_t *res = NULL; |
| axis2_char_t *id = NULL; |
| |
| savan_registry_subs_mgr_t *subsmgrimpl = NULL; |
| subsmgrimpl = SAVAN_INTF_TO_IMPL(subsmgr); |
| |
| AXIS2_LOG_TRACE(env->log, AXIS2_LOG_SI, "[savan] Entry:savan_registry_subs_mgr_init_resource"); |
| |
| subsmgrimpl->remote_registry = remote_registry_create(env, subsmgrimpl->reg_url, |
| subsmgrimpl->username, subsmgrimpl->password); |
| if(!subsmgrimpl->remote_registry) |
| { |
| AXIS2_HANDLE_ERROR(env, SAVAN_ERROR_DATABASE_CREATION_ERROR, AXIS2_FAILURE); |
| AXIS2_LOG_ERROR(env->log, AXIS2_LOG_SI, |
| "[savan] Creating remote registry instance failed"); |
| return AXIS2_FAILURE; |
| } |
| |
| res = remote_registry_get(subsmgrimpl->remote_registry, env, TOPIC_INDEX, NULL); |
| if(!res) |
| { |
| res = remote_registry_resource_create(env); |
| id = axutil_strcat(env, subsmgrimpl->reg_url, TOPIC_INDEX, NULL); |
| remote_registry_resource_set_properties(res, env, axutil_hash_make(env)); |
| remote_registry_resource_set_description(res, env, ""); |
| remote_registry_put(subsmgrimpl->remote_registry, env, TOPIC_INDEX, res); |
| } |
| |
| if(id) |
| { |
| AXIS2_FREE(env->allocator, id); |
| } |
| if(res) |
| { |
| remote_registry_resource_free(res, env); |
| } |
| |
| AXIS2_LOG_TRACE(env->log, AXIS2_LOG_SI, "[savan] Exit:savan_registry_subs_mgr_init_resource"); |
| |
| return AXIS2_SUCCESS; |
| } |
| |
| static axis2_char_t *savan_registry_subs_mgr_serialize_endpoint( |
| const axutil_env_t *env, |
| const savan_subscriber_t *subscriber) |
| { |
| axiom_namespace_t *ns = NULL; |
| axiom_node_t *subs_node = NULL; |
| axiom_element_t *subs_elem = NULL; |
| axiom_node_t *endpoint_node = NULL; |
| axiom_element_t *endpoint_elem = NULL; |
| axiom_node_t *addr_node = NULL; |
| axiom_element_t *addr_elem = NULL; |
| axiom_attribute_t *url_attr = NULL; |
| axis2_char_t *notifyto = NULL; |
| char *content = NULL; |
| |
| AXIS2_LOG_TRACE(env->log, AXIS2_LOG_SI, |
| "[savan] Entry:savan_registry_subs_mgr_serialize_endpoint"); |
| |
| /* Format of the message is as |
| * <subscription><syn:endpoint xmlns:syn="http://ws.apache.org/ns/synapse"><syn:address uri= |
| * "http://localhost:9000/services/SimpleStockQuoteService" /></syn:endpoint></subscription> |
| */ |
| ns = axiom_namespace_create (env, SYNAPSE_NAMESPACE, SYNAPSE_NS_PREFIX); |
| if(subscriber) |
| { |
| axis2_endpoint_ref_t *notifyto_epr = NULL; |
| |
| notifyto_epr = savan_subscriber_get_notify_to((savan_subscriber_t *) subscriber, env); |
| if(notifyto_epr) |
| { |
| notifyto = (axis2_char_t *) axis2_endpoint_ref_get_address(notifyto_epr, env); |
| } |
| } |
| |
| subs_elem = axiom_element_create(env, NULL, ELEM_NAME_SUBSCRIPTION, NULL, &subs_node); |
| endpoint_elem = axiom_element_create(env, subs_node, ELEM_NAME_ENDPOINT, ns, &endpoint_node); |
| addr_elem = axiom_element_create(env, endpoint_node, ELEM_NAME_ADDRESS, ns, &addr_node); |
| url_attr = axiom_attribute_create(env, ATTR_NAME_URL, notifyto, NULL); |
| axiom_element_add_attribute(addr_elem, env, url_attr, addr_node); |
| |
| content = (char *) axiom_node_to_string(subs_node, env); |
| AXIS2_LOG_TRACE(env->log, AXIS2_LOG_SI, |
| "[savan] Exit:savan_registry_subs_mgr_serialize_endpoint"); |
| |
| return content; |
| } |
| |
| static savan_subscriber_t *savan_registry_subs_mgr_extract_subscriber( |
| const axutil_env_t *env, |
| remote_registry_resource_t *resource) |
| { |
| savan_subscriber_t *subscriber = NULL; |
| axis2_char_t *content = NULL; |
| axutil_qname_t *qname = NULL; |
| axiom_node_t *subs_node = NULL; |
| axiom_element_t *subs_element = NULL; |
| axiom_node_t *endpoint_node = NULL; |
| axiom_element_t *endpoint_element = NULL; |
| axiom_node_t *address_node = NULL; |
| axiom_element_t *address_element = NULL; |
| axis2_char_t *address = NULL; |
| axutil_hash_t *properties = NULL; |
| axis2_char_t *static_flag = NULL; |
| axis2_char_t *subs_mgr_uri = NULL; |
| axis2_char_t *filter = NULL; |
| axis2_char_t *filter_dialect = NULL; |
| axis2_char_t *expires = NULL; |
| axis2_endpoint_ref_t *notifyto_epr = NULL; |
| |
| content = remote_registry_resource_get_content(resource, env); |
| subs_node = axiom_node_create_from_buffer(env, content); |
| subs_element = axiom_node_get_data_element(subs_node, env); |
| if(!subs_element) |
| { |
| AXIS2_HANDLE_ERROR(env, AXIS2_ERROR_OM_ELEMENT_EXPECTED, AXIS2_FAILURE); |
| return NULL; |
| } |
| |
| qname = axutil_qname_create(env, ELEM_NAME_ENDPOINT, SYNAPSE_NAMESPACE, NULL); |
| if(!qname) |
| { |
| AXIS2_HANDLE_ERROR(env, AXIS2_ERROR_NO_MEMORY, AXIS2_FAILURE); |
| return NULL; |
| } |
| |
| endpoint_element = axiom_element_get_first_child_with_qname(subs_element, env, qname, |
| subs_node, &endpoint_node); |
| |
| if(qname) |
| { |
| axutil_qname_free(qname, env); |
| } |
| |
| qname = axutil_qname_create(env, ELEM_NAME_ADDRESS, SYNAPSE_NAMESPACE, NULL); |
| if(!qname) |
| { |
| AXIS2_HANDLE_ERROR(env, AXIS2_ERROR_NO_MEMORY, AXIS2_FAILURE); |
| return NULL; |
| } |
| |
| address_element = axiom_element_get_first_child_with_qname(endpoint_element, env, qname, |
| endpoint_node, &address_node); |
| |
| if(qname) |
| { |
| axutil_qname_free(qname, env); |
| } |
| |
| address = axiom_element_get_attribute_value_by_name(address_element, env, ATTR_NAME_URL); |
| |
| if(!address) |
| { |
| AXIS2_HANDLE_ERROR(env, AXIS2_ERROR_OM_ELEMENT_INVALID_STATE, AXIS2_FAILURE); |
| return NULL; |
| } |
| |
| notifyto_epr = axis2_endpoint_ref_create(env, address); |
| if(!notifyto_epr) |
| { |
| AXIS2_HANDLE_ERROR(env, AXIS2_ERROR_NO_MEMORY, AXIS2_FAILURE); |
| return NULL; |
| } |
| |
| subscriber = savan_subscriber_create(env); |
| if(!subscriber) |
| { |
| return NULL; |
| } |
| |
| savan_subscriber_set_notify_to(subscriber, env, notifyto_epr); |
| |
| properties = remote_registry_resource_get_properties(resource, env); |
| if(!properties) |
| { |
| savan_subscriber_free(subscriber, env); |
| AXIS2_HANDLE_ERROR(env, SAVAN_ERROR_PARSING_SUBSCRIBER_NODE_FAILED, AXIS2_FAILURE); |
| return NULL; |
| } |
| |
| expires = remote_registry_resource_get_property(resource, env, "expires"); |
| if(expires) |
| { |
| savan_subscriber_set_expires(subscriber, env, expires); |
| } |
| |
| static_flag = remote_registry_resource_get_property(resource, env, "staticFlag"); |
| filter = remote_registry_resource_get_property(resource, env, "filterValue"); |
| if(filter) |
| { |
| savan_subscriber_set_filter(subscriber, env, filter); |
| } |
| |
| filter_dialect = remote_registry_resource_get_property(resource, env, "filterDialect"); |
| if(filter_dialect) |
| { |
| savan_subscriber_set_filter_dialect(subscriber, env, filter_dialect); |
| } |
| |
| subs_mgr_uri = remote_registry_resource_get_property(resource, env, "subManagerURI"); |
| |
| return subscriber; |
| } |
| |
