1		/*
2		* Copyright 1999-2006 University of Chicago
3		*
4		* Licensed under the Apache License, Version 2.0 (the "License");
5		* you may not use this file except in compliance with the License.
6		* You may obtain a copy of the License at
7		*
8		* http://www.apache.org/licenses/LICENSE-2.0
9		*
10		* Unless required by applicable law or agreed to in writing, software
11		* distributed under the License is distributed on an "AS IS" BASIS,
12		* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13		* See the License for the specific language governing permissions and
14		* limitations under the License.
15		*/
16
17		#include "globus_common.h"
18		#include "globus_wsrf_resource.h"
19		#include "globus_service_engine.h"
20		#include "globus_service_registry.h"
21		#include "globus_wsrf_core_tools.h"
22		#include "wsnt_NotifyType.h"
23		#include "globus_i_notification_consumer.h"
24		#include "NotificationConsumerService.h"
25		#include "globus_xml_buffer.h"
26		#include "wsnt_NotificationMessageHolderType.h"
27
28		#ifndef GLOBUS_DONT_DOCUMENT_INTERNAL
29		/**
30		* Topic expressions we know to be string valued. Anything else we
31		* leave alone.
32		*/
33		/* @{ */
34		#define SIMPLE_TOPIC_EXPRESSION \
35		"http://docs.oasis-open.org/wsn/2004/06/TopicExpression/Simple"
36		#define CONCRETE_TOPIC_EXPRESSION \
37		"http://docs.oasis-open.org/wsn/2004/06/TopicExpression/Concrete"
38		#define FULL_TOPIC_EXPRESSION \
39		"http://docs.oasis-open.org/wsn/2004/06/TopicExpression/Full"
40		/* @} */
41
42		static
43		globus_result_t
44		globus_i_notification_consumer_service_deserialize_topic(
45		wsnt_TopicExpressionType *          topic);
46
47		/* Implementation of ConsumerService skeleton functions */
48		extern
49		globus_result_t
50		NotificationConsumerService_init(
51		globus_service_descriptor_t *       service_desc)
52	8	{
53	8	globus_module_activate(GLOBUS_NOTIFICATION_CONSUMER_MODULE);
54
55	8	return GLOBUS_SUCCESS;
56		}
57		/* NotificationConsumerService_init() */
58
59		extern
60		globus_result_t
61		NotificationConsumerService_finalize(
62		globus_service_descriptor_t *       service_desc)
63	0	{
64	0	globus_module_deactivate(GLOBUS_NOTIFICATION_CONSUMER_MODULE);
65
66	0	return GLOBUS_SUCCESS;
67		}
68		/* NotificationConsumerService_finalize() */
69
70		globus_result_t
71		Consumer_Notify_impl(
72		globus_service_engine_t             engine,
73		globus_soap_message_handle_t        message,
74		globus_service_descriptor_t *       service,
75		wsnt_NotifyType *                   Notify)
76	30	{
77	30	globus_result_t                     result;
78	30	globus_resource_t                   resource;
79	30	globus_i_notification_consumer_t *  consumer;
80	30	int                                 i;
81
82	30	globus_mutex_lock(&globus_i_notification_lock);
83
84	30	result = globus_wsrf_core_get_resource(message, service, &resource);
85
86	30	if (result != GLOBUS_SUCCESS || resource == NULL)
87		{
88	0	result = GLOBUS_NOTIFICATION_CONSUMER_ERROR_UNKNOWN_RESOURCE(
89		result ? globus_error_get(result) : NULL);
90
91	0	goto unlock_out;
92		}
93
94	30	result = globus_resource_get_resource_specific(
95		resource,
96		(void **)&consumer);
97
98	30	if (result != GLOBUS_SUCCESS || consumer == NULL)
99		{
100	0	result = GLOBUS_NOTIFICATION_CONSUMER_ERROR_UNKNOWN_RESOURCE(
101		result ? globus_error_get(result) : NULL);
102
103	0	goto finish_out;
104		}
105
106	60	for (i = 0; i < Notify->NotificationMessage.length; i++)
107		{
108	30	globus_bool_t                   called = GLOBUS_FALSE;
109		wsnt_NotificationMessageHolderType *
110	30	input = &Notify->NotificationMessage.elements[i];
111
112	30	result = globus_i_notification_consumer_service_deserialize_topic(
113		&input->Topic);
114
115	30	if (result != GLOBUS_SUCCESS)
116		{
117	0	continue;
118		}
119
120
121	60	while (!called)
122		{
123	30	switch (consumer->state)
124		{
125		case CONSUMER_OK:
126	30	consumer->state = CONSUMER_ACTIVE;
127	30	globus_mutex_unlock(&globus_i_notification_lock);
128
129
130	30	consumer->callback(
131		consumer->arg,
132		&input->Topic,
133		input->ProducerReference,
134		&input->Message);
135
136	30	globus_mutex_lock(&globus_i_notification_lock);
137
138	30	if (consumer->state == CONSUMER_ACTIVE)
139		{
140	30	consumer->state = CONSUMER_OK;
141		}
142	0	else if (consumer->state == CONSUMER_ACTIVE_DESTROY_CALLED)
143		{
144	0	consumer->state = CONSUMER_DESTROYED;
145		}
146	30	called = GLOBUS_TRUE;
147	30	globus_cond_broadcast(&globus_i_notification_cond);
148
149	30	break;
150		case CONSUMER_ACTIVE:
151	0	globus_cond_wait(
152		&globus_i_notification_cond,
153		&globus_i_notification_lock);
154	0	break;
155		case CONSUMER_ACTIVE_DESTROY_CALLED:
156		case CONSUMER_DESTROYED:
157		/* Can't call it now */
158	0	called = GLOBUS_TRUE;
159	0	break;
160		}
161		}
162		}
163
164		finish_out:
165	30	result = globus_resource_finish(resource);
166
167		unlock_out:
168	30	globus_mutex_unlock(&globus_i_notification_lock);
169	30	return result;
170		}
171		/* Consumer_Notify_impl() */
172
173		/**
174		* Deserialize the topic expression into an appropriate xsd type if the
175		* dialect is known to us. Otherwise, leave it alone.
176		*
177		* @param topic
178		*     TopicExpression which will be potentially modified with a new
179		*     value of any if the Dialect attribute refers to one that this
180		*     function knows of.
181		*/
182		static
183		globus_result_t
184		globus_i_notification_consumer_service_deserialize_topic(
185		wsnt_TopicExpressionType *          topic)
186	30	{
187	30	xsd_string *                        topic_string = NULL;
188	30	globus_xml_buffer *                 buffer;
189	30	globus_result_t                     result = GLOBUS_SUCCESS;
190	30	globus_soap_message_handle_t        message_handle;
191
192		/* Deserialize the notification topic to a string if it is one of the
193		* known string-based topic dialects.
194		*/
195	30	if (topic->_Dialect &&
196		topic->any && topic->any->any_info == &globus_xml_buffer_info &&
197		(strcmp(*topic->_Dialect, SIMPLE_TOPIC_EXPRESSION) == 0 ||
198		strcmp(*topic->_Dialect, CONCRETE_TOPIC_EXPRESSION) == 0 ||
199		strcmp(*topic->_Dialect, FULL_TOPIC_EXPRESSION) == 0))
200		{
201	0	buffer = topic->any->value;
202
203	0	result = globus_soap_message_handle_init_from_memory(
204		&message_handle,
205		buffer->buffer,
206		buffer->length);
207
208	0	if (result != GLOBUS_SUCCESS)
209		{
210	0	goto out;
211		}
212
213	0	result = xsd_string_init(&topic_string);
214
215	0	if (result != GLOBUS_SUCCESS || topic_string == NULL)
216		{
217	0	goto destroy_handle_out;
218		}
219
220	0	result = xsd_string_deserialize_contents(
221		NULL,
222		topic_string,
223		message_handle,
224		0);
225
226	0	if (result != GLOBUS_SUCCESS)
227		{
228	0	goto destroy_topic_out;
229		}
230
231	0	globus_xml_buffer_destroy(buffer);
232
233	0	topic->any->any_info = &xsd_string_contents_info;
234	0	topic->any->value = topic_string;
235
236	0	topic_string = NULL;
237		}
238		else
239		{
240		/* Don't know this dialect, leave it as an xml buffer */
241	0	goto out;
242		}
243
244
245		destroy_topic_out:
246	0	if (topic_string != NULL)
247		{
248	0	xsd_string_destroy(topic_string);
249		}
250		destroy_handle_out:
251	0	globus_soap_message_handle_destroy(message_handle);
252		out:
253	30	if (result != GLOBUS_SUCCESS)
254		{
255	0	return GLOBUS_NOTIFICATION_CONSUMER_ERROR_DESERIALIZING_TOPIC(
256		globus_error_get(result));
257		}
258		else
259		{
260	30	return result;
261		}
262		}
263		/* globus_i_notification_consumer_service_deserialize_topic() */