http://git-wip-us.apache.org/repos/asf/airavata/blob/cab15715/modules/ws-messenger/README.txt ---------------------------------------------------------------------- diff --git a/modules/ws-messenger/README.txt b/modules/ws-messenger/README.txt deleted file mode 100644 index 946b571..0000000 --- a/modules/ws-messenger/README.txt +++ /dev/null @@ -1,102 +0,0 @@ -WS-Messenger 0.0.1-SNAPSHOT ---------------- - - - - -New Features In This Release ----------------------------- - - - -Key Features ------------- - - -Issues Fixed in This Release ----------------------------- - - -Installation & Running ----------------------- -1. extract the downloaded zip file -2. Run the axis2server.sh or axis2server.bat file in the bin directory -3. Once the server starts, point your Web browser to - https://localhost:8280/axis2/services -4. You can see the deployed services in simple Axis2 server, run the samples in client-api/samples directory. - -For more details, see the Installation Guide - -System Requirements -------------------- - - - -Known Issues in This Release ----------------------------- - - - -Including External Dependencies --------------------------------- - - -WS-Messenger Binary Distribution Directory Structure --------------------------------------------- - - WS-MESSENGER_HOME - |-- gui <folder> - |-- standalone-server <folder> - | |-- lib <folder> - | |-- conf <folder> - | |-- repository <folder> - | -- services <folder> - | |-- bin <folder> - |-- database_scripts - |-- client-api <folder> - | |-- lib <folder> - | |-- samples <folder> - |-- LICENSE.txt <file> - |-- README.txt <file> - |-- INSTALL.txt <file> - - - - - gui - Contains the scripts to run WS Notification Listener GUI tool - - - standalone-server - This contains all the artifacts required during axis2Server runtime with required axis2 service archives - - - standalone-server - lib - Contains all the libraries required for Axis2 runtime - - - standalone-server - conf - Contains all the configuration files for Axis2 Rutime and messenger services (messagebroker and messagebox) - - - standalone-server - repository - services - Contains deployed services in Axis2 runtime. - - - standalone-server - bin - Commandline scripts to use to start the SimpleAxis2Server and other Axis2 related scripts. - - - standalone-server - bin - database_scripts - Contains the database scripts which are used to create tables for messagebox and messagebroker services - - - client-api - Contains all the client side artifacts, which contains client samples and required libraries. - - - client-api - lib - Required libraries for client side usage. - - - client-api - samples - Contains set of samples which demostrate the funcationality of messagebroker and messagebox. - - - README.txt - This document. - - - INSTALL.txt - This document will contain information on installing Airavata-WS-Messenger. - - -
http://git-wip-us.apache.org/repos/asf/airavata/blob/cab15715/modules/ws-messenger/client/pom.xml ---------------------------------------------------------------------- diff --git a/modules/ws-messenger/client/pom.xml b/modules/ws-messenger/client/pom.xml deleted file mode 100644 index dd6ce8c..0000000 --- a/modules/ws-messenger/client/pom.xml +++ /dev/null @@ -1,60 +0,0 @@ -<?xml version="1.0" encoding="UTF-8"?> - -<!--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. --> - -<project xmlns="http://maven.apache.org/POM/4.0.0"; xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd";> - - <parent> - <groupId>org.apache.airavata</groupId> - <artifactId>airavata-ws-messenger</artifactId> - <version>0.15-SNAPSHOT</version> - </parent> - - <modelVersion>4.0.0</modelVersion> - <artifactId>airavata-messenger-client</artifactId> - <name>Airavata Messenger Client</name> - <url>http://airavata.apache.org/</url> - <packaging>jar</packaging> - - <dependencies> - <!-- common --> - <dependency> - <groupId>org.apache.airavata</groupId> - <artifactId>airavata-messenger-commons</artifactId> - <version>${project.version}</version> - </dependency> - - <!-- Logging --> - <dependency> - <groupId>org.slf4j</groupId> - <artifactId>slf4j-api</artifactId> - </dependency> - - <!-- Axis2 --> - <dependency> - <groupId>org.apache.axis2</groupId> - <artifactId>axis2-transport-http</artifactId> - </dependency> - - - <!-- Testing --> - <dependency> - <groupId>junit</groupId> - <artifactId>junit</artifactId> - <scope>test</scope> - </dependency> - - <!-- RabbitMQ client --> - <dependency> - <groupId>com.rabbitmq</groupId> - <artifactId>amqp-client</artifactId> - <version>3.1.2</version> - </dependency> - </dependencies> -</project> http://git-wip-us.apache.org/repos/asf/airavata/blob/cab15715/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/CommonMsgBrokerClient.java ---------------------------------------------------------------------- diff --git a/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/CommonMsgBrokerClient.java b/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/CommonMsgBrokerClient.java deleted file mode 100644 index 8632dfc..0000000 --- a/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/CommonMsgBrokerClient.java +++ /dev/null @@ -1,157 +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. - * - */ -package org.apache.airavata.wsmg.client; - -import org.apache.airavata.wsmg.client.msgbox.MessagePuller; -import org.apache.airavata.wsmg.client.msgbox.MsgboxHandler; -import org.apache.airavata.wsmg.commons.MsgBoxQNameConstants; -import org.apache.airavata.wsmg.commons.WsmgVersion; -import org.apache.axiom.om.OMAbstractFactory; -import org.apache.axiom.om.OMFactory; -import org.apache.axiom.soap.SOAPFactory; -import org.apache.axis2.addressing.EndpointReference; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -abstract class CommonMsgBrokerClient implements MessageBrokerClient { - - protected final static OMFactory factory = OMAbstractFactory.getOMFactory(); - private final static SOAPFactory soapfactory = OMAbstractFactory.getSOAP11Factory(); - - private final static Logger logger = LoggerFactory.getLogger(CommonMsgBrokerClient.class); - protected ConsumerServer xs; - private long socketTimeout = 200000L; - - protected MsgboxHandler msgboxHandler = new MsgboxHandler(); - - - public CommonMsgBrokerClient(long timeout) { - socketTimeout = timeout; - WsmgVersion.requireVersionOrExit(WsmgVersion.getVersion()); - } - - public String[] getConsumerServiceEndpointReference() { - if (xs == null) { - throw new RuntimeException("Consumer server is not started yet"); - } - return xs.getConsumerServiceEPRs(); - } - - public void setTimeOutInMilliSeconds(long timeout) { - socketTimeout = timeout; - } - - public CommonMsgBrokerClient() { - WsmgVersion.requireVersionOrExit(WsmgVersion.getVersion()); - - } - - public long getTimeOutInMilliSeconds() { - return socketTimeout; - } - - public String subscribeMsgBox(String brokerService, EndpointReference msgBoxEpr, String topic, String xpath) - throws MsgBrokerClientException { - - String msgBoxId = null; - String msgBoxUrl = msgBoxEpr.getAddress(); - int biginIndex = msgBoxUrl.indexOf("clientid"); - msgBoxId = msgBoxUrl.substring(biginIndex + "clientid".length() + 1); - - if (msgBoxId == null) - throw new RuntimeException("Invalid Message Box EPR, message box ID is missing"); - - return subscribe(msgBoxEpr.getAddress(), topic, xpath); - } - - public String subscribeMsgBox(EndpointReference msgBoxEpr, String topicExpression, String xpathExpression, - long expireTime) throws MsgBrokerClientException { - - String msgBoxEventSink = msgBoxEpr.getAddress(); - - String formattedEventSink = null; - - if (msgBoxEpr.getAddress().contains("clientid")) { - formattedEventSink = msgBoxEventSink; - } else { - if (msgBoxEpr.getAllReferenceParameters() == null) - throw new MsgBrokerClientException("Invalid Message Box EPR, no reference parameters found"); - String msgBoxId = msgBoxEpr.getAllReferenceParameters().get(MsgBoxQNameConstants.MSG_BOXID_QNAME).getText(); - if (msgBoxId == null) - throw new MsgBrokerClientException("Invalid Message Box EPR, reference parameter MsgBoxAddr is missing"); - String format = msgBoxEventSink.endsWith("/") ? "%sclientid/%s" : "%s/clientid/%s"; - - formattedEventSink = String.format(format, msgBoxEventSink, msgBoxId); - - } - - return subscribe(new EndpointReference(formattedEventSink), topicExpression, xpathExpression, expireTime); - - } - - // ------------------------Message box user - // API----------------------------// - - public EndpointReference createPullMsgBox(String msgBoxLocation, long timeout) throws MsgBrokerClientException { - - EndpointReference ret = null; - try { - ret = msgboxHandler.createPullMsgBox(msgBoxLocation, timeout); - } catch (MsgBrokerClientException e) { - throw e; - } - - return ret; - } - - public EndpointReference createPullMsgBox(String msgBoxServerLoc) throws MsgBrokerClientException { - EndpointReference ret = null; - ret = msgboxHandler.createPullMsgBox(msgBoxServerLoc); - return ret; - } - - - public MessagePuller startPullingEventsFromMsgBox(EndpointReference msgBoxEpr, NotificationHandler handler, - long interval, long timeout) throws MsgBrokerClientException { - - MessagePuller ret = null; - ret = msgboxHandler.startPullingEventsFromMsgBox(msgBoxEpr, handler, interval, timeout); - return ret; - } - - public MessagePuller startPullingFromExistingMsgBox(EndpointReference msgBoxAddr, NotificationHandler handler, - long interval, long timeout) throws MsgBrokerClientException { - - MessagePuller ret = null; - ret = msgboxHandler.startPullingFromExistingMsgBox(msgBoxAddr, handler, interval, timeout); - return ret; - } - - public String deleteMsgBox(EndpointReference msgBoxEpr, long timeout) throws MsgBrokerClientException { - String ret = null; - ret = msgboxHandler.deleteMsgBox(msgBoxEpr, timeout); - return ret; - } - - public void stopPullingEventsFromMsgBox(MessagePuller msgPuller) { - msgboxHandler.stopPullingEventsFromMsgBox(msgPuller); - } -} http://git-wip-us.apache.org/repos/asf/airavata/blob/cab15715/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/ConsumerMsgReceiver.java ---------------------------------------------------------------------- diff --git a/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/ConsumerMsgReceiver.java b/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/ConsumerMsgReceiver.java deleted file mode 100644 index fe9c5ba..0000000 --- a/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/ConsumerMsgReceiver.java +++ /dev/null @@ -1,40 +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. - * - */ - -package org.apache.airavata.wsmg.client; - -import org.apache.axis2.AxisFault; -import org.apache.axis2.context.MessageContext; -import org.apache.axis2.engine.MessageReceiver; - -public class ConsumerMsgReceiver implements MessageReceiver { - - private ConsumerNotificationHandler notifHandler; - - public ConsumerMsgReceiver(ConsumerNotificationHandler notifHandler) { - this.notifHandler = notifHandler; - } - - public void receive(MessageContext messageCtx) throws AxisFault { - notifHandler.handleNotification(messageCtx.getEnvelope()); - } - -} http://git-wip-us.apache.org/repos/asf/airavata/blob/cab15715/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/ConsumerNotificationHandler.java ---------------------------------------------------------------------- diff --git a/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/ConsumerNotificationHandler.java b/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/ConsumerNotificationHandler.java deleted file mode 100644 index 8f60ed8..0000000 --- a/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/ConsumerNotificationHandler.java +++ /dev/null @@ -1,29 +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. - * - */ - -package org.apache.airavata.wsmg.client; - -import org.apache.axiom.soap.SOAPEnvelope; -import org.apache.axis2.AxisFault; - -public interface ConsumerNotificationHandler { - public void handleNotification(SOAPEnvelope msgEnvelope) throws AxisFault; -} http://git-wip-us.apache.org/repos/asf/airavata/blob/cab15715/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/ConsumerServer.java ---------------------------------------------------------------------- diff --git a/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/ConsumerServer.java b/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/ConsumerServer.java deleted file mode 100644 index 0f82ca6..0000000 --- a/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/ConsumerServer.java +++ /dev/null @@ -1,99 +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. - * - */ - -package org.apache.airavata.wsmg.client; - -import java.util.HashMap; -import java.util.Map; - -import org.apache.axis2.AxisFault; -import org.apache.axis2.context.ConfigurationContext; -import org.apache.axis2.context.ConfigurationContextFactory; -import org.apache.axis2.description.AxisService; -import org.apache.axis2.engine.MessageReceiver; -import org.apache.axis2.transport.http.SimpleHTTPServer; - -public class ConsumerServer { - - private static final String ADDRESSING_VALIDATE_ACTION = "addressing.validateAction"; - public static final String SUPPORT_SINGLE_OP = "supportSingleOperation"; - private ConsumerNotificationHandler handler; - private int listeningPort; - private SimpleHTTPServer server; - private AxisService consumerService = null; - - public ConsumerServer(int listenPort, ConsumerNotificationHandler h) { - - if (listenPort <= 0 || h == null) { - throw new IllegalArgumentException("invalid arguments supplied"); - } - - this.listeningPort = listenPort; - this.handler = h; - this.server = null; - this.consumerService = null; - // setConsumerServiceUrl(listenPort); - - } - - public String[] getConsumerServiceEPRs() { - return consumerService.getEPRs(); - } - - /** - * @param args - * @throws AxisFault - * @throws Exception - */ - public void start() throws AxisFault { - ConfigurationContext context = ConfigurationContextFactory.createConfigurationContextFromFileSystem(null, null); - - context.setProperty(ADDRESSING_VALIDATE_ACTION, Boolean.valueOf(false)); - - context.getAxisConfiguration().addParameter(SUPPORT_SINGLE_OP, Boolean.valueOf(true)); - - context.getAxisConfiguration().addParameter(ADDRESSING_VALIDATE_ACTION, Boolean.valueOf(false)); - - Map<String, MessageReceiver> msgRecieverMap = new HashMap<String, MessageReceiver>(); - ConsumerMsgReceiver conMsgRcv = new ConsumerMsgReceiver(handler); - - msgRecieverMap.put("http://www.w3.org/ns/wsdl/in-only";, conMsgRcv); - AxisService service = AxisService.createService(ConsumerService.class.getName(), - context.getAxisConfiguration(), msgRecieverMap, null, null, ConsumerServer.class.getClassLoader()); - - context.getAxisConfiguration().addService(service); - server = new SimpleHTTPServer(context, this.listeningPort); - server.start(); - - consumerService = service; - - } - - public void stop() { - // TODO: add an exeption - server.stop(); - } - - public int getListenPort() { - return listeningPort; - } - -} http://git-wip-us.apache.org/repos/asf/airavata/blob/cab15715/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/ConsumerServerHandler.java ---------------------------------------------------------------------- diff --git a/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/ConsumerServerHandler.java b/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/ConsumerServerHandler.java deleted file mode 100644 index 6d71961..0000000 --- a/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/ConsumerServerHandler.java +++ /dev/null @@ -1,54 +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. - * - */ - -package org.apache.airavata.wsmg.client; - -import org.apache.axis2.AxisFault; - -public class ConsumerServerHandler { - - private ConsumerServer xs = null; - - public void createConsumerServer(int listenPort, ConsumerNotificationHandler h) throws MsgBrokerClientException { - - try { - xs = new ConsumerServer(listenPort, h); - xs.start(); - } catch (AxisFault f) { - throw new MsgBrokerClientException("unable to start the consumer server", f); - } - } - - public String[] getConsumerServiceEndpointReference() throws MsgBrokerClientException { - if (xs == null) { - throw new MsgBrokerClientException("Consumer server is not started yet"); - } - return xs.getConsumerServiceEPRs(); - } - - public void shutdownConsumerService() { - if (xs != null) { - xs.stop(); - xs = null; - } - } - -} http://git-wip-us.apache.org/repos/asf/airavata/blob/cab15715/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/ConsumerService.java ---------------------------------------------------------------------- diff --git a/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/ConsumerService.java b/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/ConsumerService.java deleted file mode 100644 index 8b7fb70..0000000 --- a/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/ConsumerService.java +++ /dev/null @@ -1,30 +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. - * - */ - -package org.apache.airavata.wsmg.client; - -public class ConsumerService { - - public void consumer(String msg) { - System.out.println(msg); - } - -} http://git-wip-us.apache.org/repos/asf/airavata/blob/cab15715/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/MessageBrokerClient.java ---------------------------------------------------------------------- diff --git a/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/MessageBrokerClient.java b/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/MessageBrokerClient.java deleted file mode 100644 index c5ada0a..0000000 --- a/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/MessageBrokerClient.java +++ /dev/null @@ -1,173 +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. - * - */ - -package org.apache.airavata.wsmg.client; - -import org.apache.airavata.wsmg.client.msgbox.MessagePuller; -import org.apache.axiom.om.OMElement; -import org.apache.axis2.AxisFault; -import org.apache.axis2.addressing.EndpointReference; - -import java.rmi.RemoteException; - -public interface MessageBrokerClient { - - /** - * May be we just use a constructor here - * - * @param brokerLocation - */ - public void init(String brokerLocation); - - /** - * - * @param subscriptionId - * @return - */ - public boolean unSubscribe(String subscriptionId) throws MsgBrokerClientException; - - /** - * - * @param eventSinkLocation - * @param topicExpression - * @param xpathExpression - * @return - * @throws AxisFault - */ - public String subscribe(String eventSinkLocation, String topicExpression, String xpathExpression) - throws MsgBrokerClientException; - - public String subscribe(EndpointReference eventSinkLocation, String topicExpression, String xpathExpression) - throws MsgBrokerClientException; - - /** - * - * @param eventSinkLocation - * @param topicExpression - * @param xpathExpression - * @param expireTime - * with -1, it never expires - * @return - * @throws AxisFault - */ - public String subscribe(EndpointReference eventSinkLocation, String topicExpression, String xpathExpression, - long expireTime) throws MsgBrokerClientException; - - public void publish(String topic, String plainText) throws MsgBrokerClientException; - - public void publish(String topic, OMElement message) throws MsgBrokerClientException; - - /** - * - * @param brokerLocation - * @param msgBoxEpr - * @param topicExpression - * @param xpathExpression - * @param expireTime - * with -1, it never expires - * @return - * @throws MsgBrokerClientException - */ - public String subscribeMsgBox(EndpointReference msgBoxEpr, String topicExpression, String xpathExpression, - long expireTime) throws MsgBrokerClientException; - - /** - * This method can be used to shutdown the consumer server started - */ - public void shutdownConsumerService(); - - /** - * - * @param port - * @param handler - * @return - * @throws MsgBrokerClientException - */ - public String[] startConsumerService(int port, ConsumerNotificationHandler handler) throws MsgBrokerClientException; - - /** - * - * @param msgBoxAddr - * @param handler - * @param backoff - * @param timeout - * @return - * @throws AxisFault - */ - public MessagePuller startPullingFromExistingMsgBox(EndpointReference msgBoxAddr, NotificationHandler handler, - long backoff, long timeout) throws MsgBrokerClientException; - - /** - * - * @param msgBoxAddr - * @param handler - * @param backoff - * @param timeout - * @return - * @throws RemoteException - */ - public MessagePuller startPullingEventsFromMsgBox(EndpointReference msgBoxAddr, NotificationHandler handler, - long backoff, long timeout) throws MsgBrokerClientException; - - // public EndpointReference createPullMsgBox(String msgBoxServerLoc, String userAgent) throws RemoteException ; - - /** - * - * @param msgPuller - */ - public void stopPullingEventsFromMsgBox(org.apache.airavata.wsmg.client.msgbox.MessagePuller msgPuller) - throws MsgBrokerClientException; - - /** - * - * @return - */ - public String[] getConsumerServiceEndpointReference(); - - /** - * - * @param brokerService - * @param msgBoxEpr - * @param topic - * @param xpath - * @return - */ - public String subscribeMsgBox(String brokerService, EndpointReference msgBoxEpr, String topic, String xpath) - throws MsgBrokerClientException; - - /** - * - * @param msgBoxLocation - * @param timeout - * @return - * @throws MsgBrokerClientException - */ - public EndpointReference createPullMsgBox(String msgBoxLocation, long timeout) throws MsgBrokerClientException; - - /** - * - * @param msgBoxServerLoc - * @return - * @throws MsgBrokerClientException - */ - public EndpointReference createPullMsgBox(String msgBoxServerLoc) throws MsgBrokerClientException; - -} http://git-wip-us.apache.org/repos/asf/airavata/blob/cab15715/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/MsgBrokerClientException.java ---------------------------------------------------------------------- diff --git a/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/MsgBrokerClientException.java b/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/MsgBrokerClientException.java deleted file mode 100644 index 8529e58..0000000 --- a/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/MsgBrokerClientException.java +++ /dev/null @@ -1,37 +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. - * - */ - -package org.apache.airavata.wsmg.client; - -import org.apache.axis2.AxisFault; - -public class MsgBrokerClientException extends AxisFault { - - private static final long serialVersionUID = 5609577226544941146L; - - public MsgBrokerClientException(String msg) { - super(msg); - } - - public MsgBrokerClientException(String msg, Throwable t) { - super(msg, t); - } -} http://git-wip-us.apache.org/repos/asf/airavata/blob/cab15715/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/NotificationHandler.java ---------------------------------------------------------------------- diff --git a/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/NotificationHandler.java b/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/NotificationHandler.java deleted file mode 100644 index 0b62067..0000000 --- a/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/NotificationHandler.java +++ /dev/null @@ -1,28 +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. - * - */ - -package org.apache.airavata.wsmg.client; - -import org.apache.axis2.AxisFault; - -public interface NotificationHandler { - public void handleNotification(String message) throws AxisFault; -} http://git-wip-us.apache.org/repos/asf/airavata/blob/cab15715/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/NotificationProducerStub.java ---------------------------------------------------------------------- diff --git a/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/NotificationProducerStub.java b/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/NotificationProducerStub.java deleted file mode 100644 index 9aee0b1..0000000 --- a/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/NotificationProducerStub.java +++ /dev/null @@ -1,190 +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. - * - */ - -package org.apache.airavata.wsmg.client; - -import javax.xml.namespace.QName; - -import org.apache.airavata.wsmg.client.util.ClientUtil; -import org.apache.airavata.wsmg.commons.WsmgCommonConstants; -import org.apache.airavata.wsmg.commons.NameSpaceConstants; -import org.apache.axiom.om.OMAbstractFactory; -import org.apache.axiom.om.OMElement; -import org.apache.axiom.om.OMFactory; -import org.apache.axiom.om.util.UUIDGenerator; -import org.apache.axiom.soap.SOAPFactory; -import org.apache.axiom.soap.SOAPHeaderBlock; -import org.apache.axis2.AxisFault; -import org.apache.axis2.addressing.EndpointReference; -import org.apache.axis2.addressing.EndpointReferenceHelper; -import org.apache.axis2.client.ServiceClient; - -/** - * - * TODO To change the template for this generated type comment go to Window - - * Preferences - Java - Code Style - Code Templates - */ -/** - * This stub is NOT multi-thread safe! - */ -public class NotificationProducerStub extends WidgetStub { - private static final String RMASSERTION = "RMAssertion"; - - private static final String WSRM_NAMESPACE = "http://schemas.xmlsoap.org/ws/2005/02/rm";; - - private static final String WSRM = "wsrm"; - - private final static OMFactory factory = OMAbstractFactory.getOMFactory(); - private final static SOAPFactory soapfactory = OMAbstractFactory.getSOAP11Factory(); - - public NotificationProducerStub(EndpointReference widgetEpr, long timeout) { - super(widgetEpr, timeout); - } - - public OMElement getCurrentMessage(OMElement topicExp) throws AxisFault { - - OMElement message = factory.createOMElement("GetCurrentMessage", NameSpaceConstants.WSNT_NS); - message.addChild(topicExp); - - ServiceClient client = createServiceClient(message); - - OMElement responseMessage = client.sendReceive(message); - client.cleanupTransport(); - - if (responseMessage.getFirstElement() == null) { - return null; - } - return (OMElement) responseMessage.getChildren().next(); - } - - private ServiceClient createServiceClient(OMElement message) throws AxisFault { - - String uuid = UUIDGenerator.getUUID(); - ServiceClient client = new ServiceClient(); - - if (client.getAxisConfiguration().getModule(WsmgCommonConstants.AXIS_MODULE_NAME_ADDRESSING) != null) { - client.engageModule(WsmgCommonConstants.AXIS_MODULE_NAME_ADDRESSING); - } else { - SOAPHeaderBlock msgId = soapfactory.createSOAPHeaderBlock("MessageID", NameSpaceConstants.WSA_NS); - msgId.setText(uuid); - SOAPHeaderBlock to = soapfactory.createSOAPHeaderBlock("To", NameSpaceConstants.WSA_NS); - to.setText(opts.getTo().getAddress()); - - SOAPHeaderBlock action = soapfactory.createSOAPHeaderBlock("Action", NameSpaceConstants.WSA_NS); - action.setText(message.getNamespace().getNamespaceURI() + "/" + message.getLocalName()); - - client.addHeader(action); - client.addHeader(msgId); - client.addHeader(to); - } - - opts.setMessageId(uuid); - opts.setAction(message.getNamespace().getNamespaceURI() + "/" + message.getLocalName()); - - opts.setProperty(org.apache.axis2.transport.http.HTTPConstants.CHUNKED, Boolean.FALSE); - opts.setTimeOutInMilliSeconds(getTimeoutInMilliSeconds()); - client.setOptions(opts); - - return client; - } - - public SubscriptionStub subscribe(EndpointReference consumerReference, OMElement topicExpression, boolean useNotfy) - throws AxisFault { - return this.subscribe(consumerReference, topicExpression, null, useNotfy, false); - - } - - public SubscriptionStub subscribe(EndpointReference consumerReference, OMElement topicExpression, - OMElement xpathExpression, boolean useNotfy) throws AxisFault { - return this.subscribe(consumerReference, topicExpression, xpathExpression, useNotfy, false); - - } - - public SubscriptionStub subscribe(EndpointReference consumerReference, OMElement topicExpression, boolean useNotfy, - boolean wsrm) throws AxisFault { - return subscribe(consumerReference, topicExpression, null, useNotfy, wsrm); - } - - public SubscriptionStub subscribe(EndpointReference consumerReference, OMElement topicExpression, - OMElement xpathExpression, boolean useNotfy, boolean wsrm) throws AxisFault { - - OMElement message = factory.createOMElement("SubscribeRequest", NameSpaceConstants.WSNT_NS); - - message.declareNamespace(NameSpaceConstants.WSNT_NS); - - OMElement eprCrEl = EndpointReferenceHelper.toOM(factory, consumerReference, new QName("ConsumerReference"), - NameSpaceConstants.WSA_NS.getNamespaceURI()); - - message.addChild(eprCrEl); - eprCrEl.setNamespace(message.getNamespace()); - - if (topicExpression != null) { - message.addChild(topicExpression); - topicExpression.setNamespace(message.getNamespace()); - } - if (xpathExpression != null) { - message.addChild(xpathExpression); - xpathExpression.setNamespace(message.getNamespace()); - } - OMElement useNotifyEl = factory.createOMElement("UseNotify", message.getNamespace(), message); - - useNotifyEl.setText(useNotfy ? "true" : "false"); - - if (wsrm) { - - setPolicyAttachment(message); - } - - ServiceClient client = createServiceClient(message); - - OMElement responseMessage = client.sendReceive(message); - client.cleanupTransport(); - - OMElement sr = responseMessage.getFirstChildWithName(new QName(NameSpaceConstants.WSNT_NS.getNamespaceURI(), - "SubscriptionReference")); - - if (sr == null) { - throw new AxisFault("unable to subscribe, invalid response returned by broker"); - } - - return new SubscriptionStub(EndpointReferenceHelper.fromOM(sr), getTimeoutInMilliSeconds()); - } - - public static void verbose(String msg) { - - System.err.println(msg); - } - - protected void setPolicyAttachment(OMElement message) { - - OMElement policy = factory.createOMElement(WsmgCommonConstants.SUBSCRIPTION_POLICY, message.getNamespace(), - message); - - // builder.newFragment(WsmgConstants.WSNT_NS, - // WsmgCommonConstants.SUBSCRIPTION_POLICY); - OMElement childEl = factory.createOMElement(new QName(WSRM_NAMESPACE, WSRM)); - // XmlNamespace wsrmNamespace = builder.newNamespace(WSRM, - // WSRM_NAMESPACE); - policy.addChild(childEl); - - } - -} http://git-wip-us.apache.org/repos/asf/airavata/blob/cab15715/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/SubscriptionInfo.java ---------------------------------------------------------------------- diff --git a/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/SubscriptionInfo.java b/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/SubscriptionInfo.java deleted file mode 100644 index 7f53021..0000000 --- a/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/SubscriptionInfo.java +++ /dev/null @@ -1,161 +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. - * - */ - -package org.apache.airavata.wsmg.client; - -import org.apache.airavata.wsmg.client.util.ClientUtil; - -public class SubscriptionInfo { - String eventSourceLocIn = null; - String eventSinkLocIn = null; - String topicExpression = null; - String xpathExpression = null; - String eventSinkEndpointReferenceNS = null; - String eventSinkEndpointReference = null; - long expireTime = ClientUtil.EXPIRE_TIME; - - /** - * @param eventSourceLocIn - * @param eventSinkLocIn - * @param topicExpression - * @param xpathExpression - * @param eventSinkEndpointReferenceNS - * @param eventSinkEndpointReference - * @param expireTime - */ - public SubscriptionInfo(String eventSourceLocIn, String eventSinkLocIn, String topicExpression, - String xpathExpression, String eventSinkEndpointReferenceNS, String eventSinkEndpointReference, - long expireTime) { - this.eventSourceLocIn = eventSourceLocIn; - this.eventSinkLocIn = eventSinkLocIn; - this.topicExpression = topicExpression; - this.xpathExpression = xpathExpression; - this.eventSinkEndpointReferenceNS = eventSinkEndpointReferenceNS; - this.eventSinkEndpointReference = eventSinkEndpointReference; - this.expireTime = expireTime; - } - - /** - * @return Returns the eventSinkEndpointReference. - */ - public String getEventSinkEndpointReference() { - return eventSinkEndpointReference; - } - - /** - * @param eventSinkEndpointReference - * The eventSinkEndpointReference to set. - */ - public void setEventSinkEndpointReference(String eventSinkEndpointReference) { - this.eventSinkEndpointReference = eventSinkEndpointReference; - } - - /** - * @return Returns the eventSinkEndpointReferenceNS. - */ - public String getEventSinkEndpointReferenceNS() { - return eventSinkEndpointReferenceNS; - } - - /** - * @param eventSinkEndpointReferenceNS - * The eventSinkEndpointReferenceNS to set. - */ - public void setEventSinkEndpointReferenceNS(String eventSinkEndpointReferenceNS) { - this.eventSinkEndpointReferenceNS = eventSinkEndpointReferenceNS; - } - - /** - * @return Returns the eventSinkLocIn. - */ - public String getEventSinkLocIn() { - return eventSinkLocIn; - } - - /** - * @param eventSinkLocIn - * The eventSinkLocIn to set. - */ - public void setEventSinkLocIn(String eventSinkLocIn) { - this.eventSinkLocIn = eventSinkLocIn; - } - - /** - * @return Returns the eventSourceLocIn. - */ - public String getEventSourceLocIn() { - return eventSourceLocIn; - } - - /** - * @param eventSourceLocIn - * The eventSourceLocIn to set. - */ - public void setEventSourceLocIn(String eventSourceLocIn) { - this.eventSourceLocIn = eventSourceLocIn; - } - - /** - * @return Returns the expireTime. - */ - public long getExpireTime() { - return expireTime; - } - - /** - * @param expireTime - * The expireTime to set. - */ - public void setExpireTime(long expireTime) { - this.expireTime = expireTime; - } - - /** - * @return Returns the topicExpression. - */ - public String getTopicExpression() { - return topicExpression; - } - - /** - * @param topicExpression - * The topicExpression to set. - */ - public void setTopicExpression(String topicExpression) { - this.topicExpression = topicExpression; - } - - /** - * @return Returns the xpathExpression. - */ - public String getXpathExpression() { - return xpathExpression; - } - - /** - * @param xpathExpression - * The xpathExpression to set. - */ - public void setXpathExpression(String xpathExpression) { - this.xpathExpression = xpathExpression; - } - -} http://git-wip-us.apache.org/repos/asf/airavata/blob/cab15715/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/SubscriptionStub.java ---------------------------------------------------------------------- diff --git a/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/SubscriptionStub.java b/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/SubscriptionStub.java deleted file mode 100644 index d094d5c..0000000 --- a/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/SubscriptionStub.java +++ /dev/null @@ -1,106 +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. - * - */ - -package org.apache.airavata.wsmg.client; - -import org.apache.airavata.wsmg.commons.WsmgCommonConstants; -import org.apache.airavata.wsmg.commons.NameSpaceConstants; -import org.apache.axiom.om.OMAbstractFactory; -import org.apache.axiom.om.OMElement; -import org.apache.axiom.om.OMFactory; -import org.apache.axiom.om.util.UUIDGenerator; -import org.apache.axiom.soap.SOAPFactory; -import org.apache.axiom.soap.SOAPHeaderBlock; -import org.apache.axis2.AxisFault; -import org.apache.axis2.addressing.EndpointReference; -import org.apache.axis2.client.ServiceClient; - -/** - * - * TODO To change the template for this generated type comment go to Window - Preferences - Java - Code Style - Code - * Templates - */ -public class SubscriptionStub extends WsrfResourceStub { - - private final static OMFactory factory = OMAbstractFactory.getOMFactory(); - private final static SOAPFactory soapfactory = OMAbstractFactory.getSOAP11Factory(); - - // private final static XmlInfosetBuilder builder = XmlConstants.BUILDER; - - SubscriptionStub(EndpointReference widgetEpr, long timeout) { - super(widgetEpr, timeout); - } - - public void pauseSubscription() throws AxisFault { - - OMElement message = factory.createOMElement("PauseSubscription", NameSpaceConstants.WSNT_NS); - - ServiceClient client = createServiceClient(message); - client.sendReceive(message); - client.cleanupTransport(); - } - - public void resumeSubscription() throws AxisFault { - - OMElement message = factory.createOMElement("ResumeSubscription", NameSpaceConstants.WSNT_NS); - - ServiceClient client = createServiceClient(message); - client.sendReceive(message); - client.cleanupTransport(); - - } - - private ServiceClient createServiceClient(OMElement message) throws AxisFault { - - String uuid = UUIDGenerator.getUUID(); - ServiceClient client = new ServiceClient(); - - if (client.getAxisConfiguration().getModule(WsmgCommonConstants.AXIS_MODULE_NAME_ADDRESSING) != null) { - client.engageModule(WsmgCommonConstants.AXIS_MODULE_NAME_ADDRESSING); - } else { - SOAPHeaderBlock msgId = soapfactory.createSOAPHeaderBlock("MessageID", NameSpaceConstants.WSA_NS); - msgId.setText(uuid); - - SOAPHeaderBlock to = soapfactory.createSOAPHeaderBlock("To", NameSpaceConstants.WSA_NS); - to.setText(opts.getTo().getAddress()); - - SOAPHeaderBlock action = soapfactory.createSOAPHeaderBlock("Action", NameSpaceConstants.WSA_NS); - action.setText(message.getNamespace().getNamespaceURI() + "/" + message.getLocalName()); - - client.addHeader(action); - client.addHeader(msgId); - client.addHeader(to); - } - opts.setProperty(org.apache.axis2.transport.http.HTTPConstants.CHUNKED, Boolean.FALSE); - opts.setMessageId(uuid); - opts.setAction(message.getNamespace().getNamespaceURI() + "/" + message.getLocalName()); - opts.setTimeOutInMilliSeconds(getTimeoutInMilliSeconds()); - client.setOptions(opts); - return client; - } - - public static void verbose(String msg) { - - System.err.println(msg); - - } - -} http://git-wip-us.apache.org/repos/asf/airavata/blob/cab15715/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/WidgetStub.java ---------------------------------------------------------------------- diff --git a/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/WidgetStub.java b/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/WidgetStub.java deleted file mode 100644 index 3400390..0000000 --- a/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/WidgetStub.java +++ /dev/null @@ -1,88 +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. - * - */ - -package org.apache.airavata.wsmg.client; - -import org.apache.airavata.wsmg.commons.WsmgCommonConstants; -import org.apache.airavata.wsmg.commons.NameSpaceConstants; -import org.apache.axiom.om.OMAbstractFactory; -import org.apache.axiom.om.OMElement; -import org.apache.axiom.om.OMFactory; -import org.apache.axiom.om.util.UUIDGenerator; -import org.apache.axiom.soap.SOAPFactory; -import org.apache.axiom.soap.SOAPHeaderBlock; -import org.apache.axis2.AxisFault; -import org.apache.axis2.addressing.EndpointReference; -import org.apache.axis2.client.ServiceClient; - -/** - * This stub is NOT multi-thread safe! - */ -public class WidgetStub extends WsrfResourceStub { - - private final static OMFactory factory = OMAbstractFactory.getOMFactory(); - private final static SOAPFactory soapfactory = OMAbstractFactory.getSOAP11Factory(); - - public WidgetStub(EndpointReference widgetEpr, long timeout) { - super(widgetEpr, timeout); - } - - public void generateNotification() throws AxisFault { - - OMElement message = factory.createOMElement("generateNotification", NameSpaceConstants.WIDGET_NS); - - ServiceClient client = createServiceClient(message); - client.sendReceive(message); - client.cleanupTransport(); - - } - - private ServiceClient createServiceClient(OMElement message) throws AxisFault { - - ServiceClient client = new ServiceClient(); - - String uuid = UUIDGenerator.getUUID(); - if (client.getAxisConfiguration().getModule(WsmgCommonConstants.AXIS_MODULE_NAME_ADDRESSING) != null) { - client.engageModule(WsmgCommonConstants.AXIS_MODULE_NAME_ADDRESSING); - } else { - - SOAPHeaderBlock msgId = soapfactory.createSOAPHeaderBlock("MessageID", NameSpaceConstants.WSA_NS); - msgId.setText(uuid); - - SOAPHeaderBlock to = soapfactory.createSOAPHeaderBlock("To", NameSpaceConstants.WSA_NS); - to.setText(opts.getTo().getAddress()); - - SOAPHeaderBlock action = soapfactory.createSOAPHeaderBlock("Action", NameSpaceConstants.WSA_NS); - action.setText(message.getNamespace().getNamespaceURI() + "/" + message.getLocalName()); - - client.addHeader(action); - client.addHeader(msgId); - client.addHeader(to); - } - opts.setProperty(org.apache.axis2.transport.http.HTTPConstants.CHUNKED, Boolean.FALSE); - opts.setMessageId(uuid); - opts.setAction(message.getNamespace().getNamespaceURI() + "/generateEvent"); - opts.setTimeOutInMilliSeconds(getTimeoutInMilliSeconds()); - client.setOptions(opts); - return client; - } - -} http://git-wip-us.apache.org/repos/asf/airavata/blob/cab15715/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/WseCreateSubscription.java ---------------------------------------------------------------------- diff --git a/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/WseCreateSubscription.java b/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/WseCreateSubscription.java deleted file mode 100644 index c10d123..0000000 --- a/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/WseCreateSubscription.java +++ /dev/null @@ -1,142 +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. - * - */ - -package org.apache.airavata.wsmg.client; - -import javax.xml.namespace.QName; - -import org.apache.airavata.wsmg.client.util.ClientUtil; -import org.apache.airavata.wsmg.commons.WsmgCommonConstants; -import org.apache.airavata.wsmg.commons.NameSpaceConstants; -import org.apache.axiom.om.OMAbstractFactory; -import org.apache.axiom.om.OMElement; -import org.apache.axiom.om.OMFactory; -import org.apache.axiom.om.util.UUIDGenerator; -import org.apache.axiom.soap.SOAPFactory; -import org.apache.axiom.soap.SOAPHeaderBlock; -import org.apache.axis2.AxisFault; -import org.apache.axis2.addressing.EndpointReference; -import org.apache.axis2.addressing.EndpointReferenceHelper; -import org.apache.axis2.client.Options; -import org.apache.axis2.client.ServiceClient; - -public class WseCreateSubscription { - private final OMFactory factory = OMAbstractFactory.getOMFactory(); - private final static SOAPFactory soapfactory = OMAbstractFactory.getSOAP11Factory(); - protected EndpointReference brokerEndPointReference; - - protected long timeoutInMilliSeconds; - - public WseCreateSubscription(EndpointReference brokerLocationEPR, long timeout) { - this.brokerEndPointReference = brokerLocationEPR; - timeoutInMilliSeconds = timeout; - } - - public long getTimeoutInMilliSeconds() { - return timeoutInMilliSeconds; - } - - public void setTimeoutInMilliSeconds(long timeout) { - timeoutInMilliSeconds = timeout; - } - - public OMElement subscribe(EndpointReference eventSinkReference, OMElement filterEl, boolean useNotfy) - throws AxisFault { - return subscribe(eventSinkReference, filterEl, useNotfy, ClientUtil.EXPIRE_TIME); - } - - public OMElement subscribe(EndpointReference eventSinkReferenceEPR, OMElement filterEl, boolean useNotfy, - long expireTime) throws AxisFault { - - OMElement message = createMessageEl(eventSinkReferenceEPR, filterEl, expireTime); - - ServiceClient serviceClient = createServiceClient(message); - OMElement responseMessage = serviceClient.sendReceive(message); - serviceClient.cleanup(); - serviceClient.cleanupTransport(); - if (responseMessage == null) { - throw AxisFault.makeFault(new RuntimeException("no response recieved for subscription message")); - } - OMElement responseSubscriptionsManagerElement = responseMessage.getFirstChildWithName(new QName( - NameSpaceConstants.WSE_NS.getNamespaceURI(), "SubscriptionManager")); - return responseSubscriptionsManagerElement; - } - - private ServiceClient createServiceClient(OMElement message) throws AxisFault { - Options opts = new Options(); - opts.setProperty(org.apache.axis2.transport.http.HTTPConstants.CHUNKED, Boolean.FALSE); - opts.setTo(brokerEndPointReference); - opts.setAction(message.getNamespace().getNamespaceURI() + "/" + message.getLocalName()); - - opts.setTimeOutInMilliSeconds(getTimeoutInMilliSeconds()); - opts.setMessageId(UUIDGenerator.getUUID()); - - ServiceClient client = new ServiceClient(); - - if (client.getAxisConfiguration().getModule(WsmgCommonConstants.AXIS_MODULE_NAME_ADDRESSING) != null) { - - client.engageModule(WsmgCommonConstants.AXIS_MODULE_NAME_ADDRESSING); - } else { - SOAPHeaderBlock msgId = soapfactory.createSOAPHeaderBlock("MessageID", NameSpaceConstants.WSA_NS); - msgId.setText(UUIDGenerator.getUUID()); - - SOAPHeaderBlock to = soapfactory.createSOAPHeaderBlock("To", NameSpaceConstants.WSA_NS); - to.setText(opts.getTo().getAddress()); - - SOAPHeaderBlock action = soapfactory.createSOAPHeaderBlock("Action", NameSpaceConstants.WSA_NS); - action.setText(message.getNamespace().getNamespaceURI() + "/" + message.getLocalName()); - - client.addHeader(action); - client.addHeader(msgId); - client.addHeader(to); - } - client.setOptions(opts); - - return client; - } - - private OMElement createMessageEl(EndpointReference eventSinkReferenceEPR, OMElement filterEl, long expireTime) - throws AxisFault { - - OMElement message = factory.createOMElement("Subscribe", NameSpaceConstants.WSE_NS); - - OMElement delivery = factory.createOMElement("Delivery", NameSpaceConstants.WSE_NS); - - OMElement expires = factory.createOMElement("Expires", NameSpaceConstants.WSE_NS); - expires.setText(Long.toString(expireTime)); - message.addChild(expires); - - OMElement notifyTo = EndpointReferenceHelper.toOM(factory, eventSinkReferenceEPR, new QName( - NameSpaceConstants.WSE_NS.getNamespaceURI(), "NotifyTo"), NameSpaceConstants.WSA_NS.getNamespaceURI()); - - delivery.addChild(notifyTo); - message.addChild(delivery); - - if (filterEl != null) { - message.addChild(filterEl); - } - - message.declareNamespace(NameSpaceConstants.WSA_NS); - - return message; - } - -} http://git-wip-us.apache.org/repos/asf/airavata/blob/cab15715/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/WseMsgBrokerClient.java ---------------------------------------------------------------------- diff --git a/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/WseMsgBrokerClient.java b/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/WseMsgBrokerClient.java deleted file mode 100644 index 5c173e8..0000000 --- a/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/WseMsgBrokerClient.java +++ /dev/null @@ -1,201 +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. - * - */ - -package org.apache.airavata.wsmg.client; - -import javax.xml.namespace.QName; - -import org.apache.airavata.wsmg.client.commons.NotificationProducer; -import org.apache.airavata.wsmg.client.protocol.WSEProtocolClient; -import org.apache.airavata.wsmg.client.util.ClientUtil; -import org.apache.airavata.wsmg.commons.CommonRoutines; -import org.apache.airavata.wsmg.commons.WsmgCommonConstants; -import org.apache.airavata.wsmg.commons.NameSpaceConstants; -import org.apache.axiom.om.OMAbstractFactory; -import org.apache.axiom.om.OMElement; -import org.apache.axiom.om.OMFactory; -import org.apache.axis2.AxisFault; -import org.apache.axis2.addressing.EndpointReference; -import org.apache.axis2.client.Options; -import org.apache.axis2.client.ServiceClient; - -public class WseMsgBrokerClient extends CommonMsgBrokerClient implements MessageBrokerClient { - - private EndpointReference brokerEndpointRef = null; - - protected long timeoutInMilliSeconds = WsmgCommonConstants.DEFAULT_CLIENT_SOCKET_TIME_OUT_MILLIES; - - final static String WIDGET_NS_PREFIX = NameSpaceConstants.WIDGET_NS.getPrefix() + ":"; - - protected NotificationProducer notificationProducer = new NotificationProducer(); - protected ConsumerServerHandler consumerServerHandler = new ConsumerServerHandler(); - - public static EndpointReference createEndpointReference(String brokerURL, String topic) { - - return WSEProtocolClient.createEndpointReference(brokerURL, topic); - } - - public void init(String brokerLocation) { - - brokerEndpointRef = new EndpointReference(ClientUtil.formatURLString(brokerLocation)); - - } - - public void publish(String topic, String plainText) throws MsgBrokerClientException { - - OMFactory factory = OMAbstractFactory.getOMFactory(); - - OMElement wrappedMsg = factory.createOMElement(WsmgCommonConstants.WSMG_PLAIN_TEXT_WRAPPER, - NameSpaceConstants.WSMG_NS); - - wrappedMsg.setText(plainText); - publish(topic, wrappedMsg); - } - - public void publish(String topic, OMElement message) throws MsgBrokerClientException { - - OMFactory factory = OMAbstractFactory.getOMFactory(); - - OMElement topicExpressionEl = null; - - if (topic != null) { - topicExpressionEl = factory.createOMElement("Topic", NameSpaceConstants.WSNT_NS); - topicExpressionEl.addAttribute("Dialect", WsmgCommonConstants.TOPIC_EXPRESSION_SIMPLE_DIALECT, null); - topicExpressionEl.declareNamespace(NameSpaceConstants.WIDGET_NS); - topicExpressionEl.setText(WIDGET_NS_PREFIX + topic); - - } - - try { - notificationProducer.deliverMessage(message, "wse", brokerEndpointRef, getTimeoutInMilliSeconds(), - topicExpressionEl); - } catch (AxisFault e) { - throw new MsgBrokerClientException("unable to publish msg", e); - } - } - - public String subscribe(String eventSinkLocation, String topicExpression, String xpathExpression) - throws MsgBrokerClientException { - return subscribe(new EndpointReference(eventSinkLocation), topicExpression, xpathExpression); - } - - public String subscribe(EndpointReference eventSinkLocation, String topicExpression, String xpathExpression) - throws MsgBrokerClientException { - - return subscribe(eventSinkLocation, topicExpression, xpathExpression, - WsmgCommonConstants.DEFAULT_SUBSCRIPTION_EXPIRATION_TIME); - } - - public String subscribe(EndpointReference eventSinkLocation, String topicExpression, String xpathExpression, - long expireTime) throws MsgBrokerClientException { - - String subscriptionId = null; - ServiceClient serviceClient = null; - try { - OMElement subscriptionMsg = WSEProtocolClient.createSubscription(eventSinkLocation, topicExpression, - xpathExpression, expireTime); - serviceClient = configureServiceClientForSubscription(subscriptionMsg); - OMElement responseMessage = serviceClient.sendReceive(subscriptionMsg); - - if (responseMessage == null) { - throw new MsgBrokerClientException("no response recieved for subscription message"); - } - - OMElement responseSubscriptionsManagerElement = responseMessage.getFirstChildWithName(new QName( - NameSpaceConstants.WSE_NS.getNamespaceURI(), "SubscriptionManager")); - - subscriptionId = WSEProtocolClient.decodeSubscribeResponse(responseSubscriptionsManagerElement); - - } catch (AxisFault e) { - throw new MsgBrokerClientException("unable to send the subscription", e); - } finally { - try { - serviceClient.cleanup(); - serviceClient.cleanupTransport(); - } catch (AxisFault e) { - throw new MsgBrokerClientException("unable to send the subscription", e); - } - } - - return subscriptionId; - } - - public boolean unSubscribe(String subscriptionId) throws MsgBrokerClientException { - OMFactory factory = OMAbstractFactory.getOMFactory(); - OMElement message = WSEProtocolClient.createUnsubscribeMsg(); - ServiceClient client = null; - OMElement identifierEl = factory.createOMElement("Identifier", message.getNamespace()); - - identifierEl.setText(subscriptionId); - String soapAction = message.getNamespace().getNamespaceURI() + "/" + message.getLocalName(); - - Options opts = CommonRoutines.getOptions(soapAction, getTimeoutInMilliSeconds(), brokerEndpointRef); - - try { - client = new ServiceClient(); - client.setOptions(opts); - CommonRoutines.setHeaders(soapAction, brokerEndpointRef.getAddress(), client, identifierEl); - client.sendReceive(message); - } catch (AxisFault e) { - throw new MsgBrokerClientException("unable to send unsubscribe msg", e); - } finally { - try { - client.cleanup(); - client.cleanupTransport(); - } catch (AxisFault e) { - throw new MsgBrokerClientException("unable to send unsubscribe msg", e); - } - } - return true; - } - - private ServiceClient configureServiceClientForSubscription(OMElement message) throws AxisFault { - - String soapAction = message.getNamespace().getNamespaceURI() + "/" + message.getLocalName(); - - Options opts = CommonRoutines.getOptions(soapAction, getTimeoutInMilliSeconds(), brokerEndpointRef); - - ServiceClient client = new ServiceClient(); - client.setOptions(opts); - CommonRoutines.setHeaders(soapAction, brokerEndpointRef.getAddress(), client); - return client; - } - - public long getTimeoutInMilliSeconds() { - return timeoutInMilliSeconds; - } - - public void setTimeoutInMilliSeconds(long timeout) { - timeoutInMilliSeconds = timeout; - } - - public String[] startConsumerService(int port, ConsumerNotificationHandler handler) throws MsgBrokerClientException { - - consumerServerHandler.createConsumerServer(port, handler); - return consumerServerHandler.getConsumerServiceEndpointReference(); - - } - - public void shutdownConsumerService() { - consumerServerHandler.shutdownConsumerService(); - } - -} http://git-wip-us.apache.org/repos/asf/airavata/blob/cab15715/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/WsntCreatePublishReg.java ---------------------------------------------------------------------- diff --git a/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/WsntCreatePublishReg.java b/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/WsntCreatePublishReg.java deleted file mode 100644 index e14ae9e..0000000 --- a/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/WsntCreatePublishReg.java +++ /dev/null @@ -1,118 +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. - * - */ - -package org.apache.airavata.wsmg.client; - -import javax.xml.namespace.QName; - -import org.apache.airavata.wsmg.commons.WsmgCommonConstants; -import org.apache.airavata.wsmg.commons.NameSpaceConstants; -import org.apache.axiom.om.OMAbstractFactory; -import org.apache.axiom.om.OMElement; -import org.apache.axiom.om.OMFactory; -import org.apache.axiom.om.util.UUIDGenerator; -import org.apache.axiom.soap.SOAPFactory; -import org.apache.axiom.soap.SOAPHeaderBlock; -import org.apache.axis2.AxisFault; -import org.apache.axis2.addressing.EndpointReference; -import org.apache.axis2.addressing.EndpointReferenceHelper; -import org.apache.axis2.client.ServiceClient; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -/** - * - * TODO To change the template for this generated type comment go to Window - Preferences - Java - Code Style - Code - * Templates - */ -public class WsntCreatePublishReg extends WidgetStub { - private static final Logger logger = LoggerFactory.getLogger(WseCreateSubscription.class); - private OMFactory factory = OMAbstractFactory.getOMFactory(); - private final static SOAPFactory soapfactory = OMAbstractFactory.getSOAP11Factory(); - - // private final static XmlInfosetBuilder builder = XmlConstants.BUILDER; - - public WsntCreatePublishReg(EndpointReference widgetEpr, long timeout) { - super(widgetEpr, timeout); - } - - public WsrfResourceStub registerPublisher(EndpointReference omEndpointReference) throws AxisFault { - - // invoker.setMessageId(null); - OMElement message = factory.createOMElement(new QName(NameSpaceConstants.WSBR_NS.getNamespaceURI(), - "RegisterPublisher")); - // builder.newFragment(WsmgConstants.WSBR_NS, - // "RegisterPublisher"); - - // EndpointReference publisherRef; - - OMElement publisherRef = null; - - publisherRef = EndpointReferenceHelper.toOM(factory, omEndpointReference, - new QName(NameSpaceConstants.WSBR_NS.getNamespaceURI(), "PublisherReference"), - NameSpaceConstants.WSA_NS.getNamespaceURI()); - - message.addChild(publisherRef); - - ServiceClient client = createServiceClient(message); - - OMElement responseMessage = client.sendReceive(message); - client.cleanup(); - client.cleanupTransport(); - - OMElement publisherRegistrationRef = responseMessage.getFirstChildWithName(new QName(NameSpaceConstants.WSBR_NS - .getNamespaceURI(), "PublisherRegistrationReference")); - - EndpointReference epr = EndpointReferenceHelper.fromOM(publisherRegistrationRef); - return new WsrfResourceStub(epr, getTimeoutInMilliSeconds()); - } - - private ServiceClient createServiceClient(OMElement message) throws AxisFault { - String uuid = UUIDGenerator.getUUID(); - opts.setProperty(org.apache.axis2.transport.http.HTTPConstants.CHUNKED, Boolean.FALSE); - opts.setMessageId(uuid); - opts.setAction(message.getNamespace().getNamespaceURI() + "/" + message.getLocalName()); - opts.setTimeOutInMilliSeconds(getTimeoutInMilliSeconds()); - - ServiceClient client = new ServiceClient(); - - if (client.getAxisConfiguration().getModule(WsmgCommonConstants.AXIS_MODULE_NAME_ADDRESSING) != null) { - - client.engageModule(WsmgCommonConstants.AXIS_MODULE_NAME_ADDRESSING); - } else { - SOAPHeaderBlock msgId = soapfactory.createSOAPHeaderBlock("MessageID", NameSpaceConstants.WSA_NS); - msgId.setText(uuid); - - SOAPHeaderBlock to = soapfactory.createSOAPHeaderBlock("To", NameSpaceConstants.WSA_NS); - to.setText(opts.getTo().getAddress()); - - SOAPHeaderBlock action = soapfactory.createSOAPHeaderBlock("Action", NameSpaceConstants.WSA_NS); - action.setText(message.getNamespace().getNamespaceURI() + "/" + message.getLocalName()); - - client.addHeader(action); - client.addHeader(msgId); - client.addHeader(to); - } - client.setOptions(opts); - return client; - } - -} http://git-wip-us.apache.org/repos/asf/airavata/blob/cab15715/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/WsntMsgBrokerClient.java ---------------------------------------------------------------------- diff --git a/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/WsntMsgBrokerClient.java b/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/WsntMsgBrokerClient.java deleted file mode 100644 index 321bfe0..0000000 --- a/modules/ws-messenger/client/src/main/java/org/apache/airavata/wsmg/client/WsntMsgBrokerClient.java +++ /dev/null @@ -1,190 +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. - * - */ - -package org.apache.airavata.wsmg.client; - -import javax.xml.namespace.QName; - -import org.apache.airavata.wsmg.client.commons.NotificationProducer; -import org.apache.airavata.wsmg.client.protocol.WSNTProtocolClient; -import org.apache.airavata.wsmg.client.util.ClientUtil; -import org.apache.airavata.wsmg.commons.CommonRoutines; -import org.apache.airavata.wsmg.commons.WsmgCommonConstants; -import org.apache.airavata.wsmg.commons.NameSpaceConstants; -import org.apache.axiom.om.OMAbstractFactory; -import org.apache.axiom.om.OMElement; -import org.apache.axiom.om.OMFactory; -import org.apache.axis2.AxisFault; -import org.apache.axis2.addressing.EndpointReference; -import org.apache.axis2.client.Options; -import org.apache.axis2.client.ServiceClient; - -public class WsntMsgBrokerClient extends CommonMsgBrokerClient implements MessageBrokerClient { - - protected long timeoutInMilliSeconds = WsmgCommonConstants.DEFAULT_CLIENT_SOCKET_TIME_OUT_MILLIES; - - protected NotificationProducer notifProducer = new NotificationProducer(); - protected ConsumerServerHandler consumerServerHandler = new ConsumerServerHandler(); - - private EndpointReference brokerEndpointRef = null; - - public static EndpointReference createEndpointReference(String brokerURL, String topic) { - // TODO : implement - return null; - } - - public void init(String brokerLocation) { - brokerEndpointRef = new EndpointReference(ClientUtil.formatURLString(brokerLocation)); - - } - - public void publish(String topic, String plainText) throws MsgBrokerClientException { - - OMFactory factory = OMAbstractFactory.getOMFactory(); - - OMElement wrappedMsg = factory.createOMElement(WsmgCommonConstants.WSMG_PLAIN_TEXT_WRAPPER, - NameSpaceConstants.WSMG_NS); - - wrappedMsg.setText(plainText); - publish(topic, wrappedMsg); - - } - - public void publish(String topic, OMElement message) throws MsgBrokerClientException { - - String producerURl = ClientUtil.formatURLString(ClientUtil.getHostIP()); - EndpointReference producerReference = new EndpointReference(producerURl); - - try { - - OMElement messageToNotify = WSNTProtocolClient.encodeNotification(topic, message, producerReference); - - notifProducer.deliverMessage(messageToNotify, "wsnt", brokerEndpointRef, getTimeoutInMilliSeconds()); - } catch (AxisFault e) { - throw new MsgBrokerClientException("unable to publish msg", e); - } - - } - - public String subscribe(String eventSinkLocation, String topicExpression, String xpathExpression) - throws MsgBrokerClientException { - - return subscribe(new EndpointReference(eventSinkLocation), topicExpression, xpathExpression); - } - - public String subscribe(EndpointReference eventSinkLocation, String topicExpression, String xpathExpression) - throws MsgBrokerClientException { - return subscribe(eventSinkLocation, topicExpression, xpathExpression, - WsmgCommonConstants.DEFAULT_SUBSCRIPTION_EXPIRATION_TIME); - } - - public String subscribe(EndpointReference eventSinkLocation, String topicExpression, String xpathExpression, - long expireTime) throws MsgBrokerClientException { - - String subscriptionId = null; - - try { - - OMElement message = WSNTProtocolClient.createSubscriptionMsg(eventSinkLocation, topicExpression, - xpathExpression); - - ServiceClient client = createServiceClient(message); - - OMElement responseMessage = client.sendReceive(message); - client.cleanup(); - client.cleanupTransport(); - - OMElement sr = responseMessage.getFirstChildWithName(new QName( - NameSpaceConstants.WSNT_NS.getNamespaceURI(), "SubscriptionReference")); - - if (sr == null) { - throw new MsgBrokerClientException("unable to subscribe, invalid response returned by broker"); - } - - subscriptionId = WSNTProtocolClient.decodeSubscriptionResponse(sr); - - } catch (AxisFault f) { - throw new MsgBrokerClientException("unable to send the subscription msg", f); - } - - return subscriptionId; - - } - - private ServiceClient createServiceClient(OMElement message) throws AxisFault { - - String soapAction = message.getNamespace().getNamespaceURI() + "/" + message.getLocalName(); - - Options opts = CommonRoutines.getOptions(soapAction, getTimeoutInMilliSeconds(), brokerEndpointRef); - - ServiceClient client = new ServiceClient(); - CommonRoutines.setHeaders(soapAction, brokerEndpointRef.getAddress(), client); - - client.setOptions(opts); - - return client; - } - - public boolean unSubscribe(String subscriptionId) throws MsgBrokerClientException { - - OMFactory factory = OMAbstractFactory.getOMFactory(); - OMElement message = WSNTProtocolClient.createUnsubscribeMsg(); - OMElement identifierEl = factory.createOMElement("Identifier", message.getNamespace()); - - identifierEl.setText(subscriptionId); - String soapAction = message.getNamespace().getNamespaceURI() + "/" + message.getLocalName(); - - Options opts = CommonRoutines.getOptions(soapAction, getTimeoutInMilliSeconds(), brokerEndpointRef); - - try { - ServiceClient client = new ServiceClient(); - client.setOptions(opts); - CommonRoutines.setHeaders(soapAction, brokerEndpointRef.getAddress(), client, identifierEl); - - client.sendReceive(message); - client.cleanupTransport(); - - } catch (AxisFault e) { - throw new MsgBrokerClientException("unable to send subscribe msg", e); - } - - return true; - } - - public long getTimeoutInMilliSeconds() { - return timeoutInMilliSeconds; - } - - public void setTimeoutInMilliSeconds(long timeout) { - timeoutInMilliSeconds = timeout; - } - - public String[] startConsumerService(int port, ConsumerNotificationHandler handler) throws MsgBrokerClientException { - - consumerServerHandler.createConsumerServer(port, handler); - return consumerServerHandler.getConsumerServiceEndpointReference(); - - } - - public void shutdownConsumerService() { - consumerServerHandler.shutdownConsumerService(); - } -}
