使用Spring BlazeDS Integration进行数据推送服务(push data)

使用Consumer(mx.messaging.Consumer)向服务端订阅消息,服务端给订阅者推送消息

第一步:整合spring和BlazeDS

这一步网上资料很多,如果找不到或找到的不行的话,可留言我再将整合spring和BlazeDS整合给大家详细的解说下

第二步:

我们先说配置

services-config.xml中加入以下内容

		<channel-definition id="my-streaming-amf"
			class="mx.messaging.channels.StreamingAMFChannel">
			<endpoint
				url="http://{server.name}:{server.port}/{context.root}/messagebroker/streamingamf"
				class="flex.messaging.endpoints.StreamingAMFEndpoint" />
			<properties>
				<idle-timeout-minutes>0</idle-timeout-minutes>
				<max-streaming-clients>100</max-streaming-clients>
				<server-to-client-heartbeat-millis>1000
				</server-to-client-heartbeat-millis>
				<user-agent-settings>
					<!--
						MSIE 5, 6, 7 default max number of permanent HTTP connections is
						2.
					-->
					<user-agent match-on="MSIE" kickstart-bytes="2048"
						max-streaming-connections-per-session="1" />
					<!-- MSIE 8 max number is 6. -->
					<user-agent match-on="MSIE 8" kickstart-bytes="2048"
						max-streaming-connections-per-session="5" />
					<!-- Firefox 1, 2 max number is 2. -->
					<user-agent match-on="Firefox" kickstart-bytes="2048"
						max-streaming-connections-per-session="1" />
					<!-- Firefox 3 max number is 6. -->
					<user-agent match-on="Firefox/3" kickstart-bytes="2048"
						max-streaming-connections-per-session="5" />
					<!-- Safari 3, 4 max number is 4. -->
					<user-agent match-on="Safari" kickstart-bytes="2048"
						max-streaming-connections-per-session="3" />
					<!-- Chrome 0, 1, 2 max number is 6. -->
					<user-agent match-on="Chrome" kickstart-bytes="2048"
						max-streaming-connections-per-session="5" />
					<!-- Opera 7, 9 max number is 4.-->
					<user-agent match-on="Opera" kickstart-bytes="2048"
						max-streaming-connections-per-session="3" />
					<!-- Opera 8 max number is 8. -->
					<user-agent match-on="Opera 8" kickstart-bytes="2048"
						max-streaming-connections-per-session="7" />
					<!-- Opera 10 max number is 8. -->
					<user-agent match-on="Opera 10" kickstart-bytes="2048"
						max-streaming-connections-per-session="7" />
				</user-agent-settings>
			</properties>
		</channel-definition>

然后messaging-config.xml中加入以下配置:

<?xml version="1.0" encoding="UTF-8"?>
<service id="message-service" class="flex.messaging.services.MessageService">

	<adapters>
		<adapter-definition id="actionscript"
			class="flex.messaging.services.messaging.adapters.ActionScriptAdapter"
			default="true" />
		<!--
			<adapter-definition id="jms"
			class="flex.messaging.services.messaging.adapters.JMSAdapter" />
		-->
	</adapters>

	<default-channels>
		<channel ref="my-streaming-amf" />
		<channel ref="my-polling-amf" />
	</default-channels>

	<destination id="testDataPush">
		<properties>
			<server>
				<allow-subtopics>true</allow-subtopics>
				<subtopic-separator>.</subtopic-separator>
			</server>
		</properties>
		<channels>
			<channel ref="my-polling-amf" />
			<channel ref="my-streaming-amf" />
		</channels>
	</destination>

</service>

然后spring bean里加入:

	<bean id="defaultMessageTemplate" class="org.springframework.flex.messaging.MessageTemplate" />

不急待会会用到它,并且这很重要

第三步 服务端代码及配置

java代码:

public class TestDataPush {
	private MessageTemplate messageTemplate;
	private AsyncMessage asyncMessage;

	public TestDataPush() {
		super();

		String clientID = UUIDUtils.createUUID();
		asyncMessage = new AsyncMessage();
		asyncMessage.setDestination("testDataPush");
		asyncMessage.setHeader("DSSubtopic", "3d");
		asyncMessage.setClientId(clientID);
		asyncMessage.setMessageId(UUIDUtils.createUUID());
	}

	public void test() {
		System.out.println("test-------------------");
		asyncMessage.setTimestamp(System.currentTimeMillis());
		asyncMessage.setBody("test");
		messageTemplate.getMessageBroker().routeMessageToService(asyncMessage, null);
	}

	public MessageTemplate getMessageTemplate() {
		return messageTemplate;
	}

	public void setMessageTemplate(MessageTemplate messageTemplate) {
		this.messageTemplate = messageTemplate;
	}

}

这个java类作为flex remote调用对象,将调用test方法时我们会向订阅者发布消息
既然我们的messagebroker都交给spring管理了,相应的messagebroker我们就应该从spring提供的模板中去得到它

这时我们将之前配置的template注入到这里

	<bean id="testPush" class="TestDataPush">
		<property name="messageTemplate" ref="defaultMessageTemplate"></property>
	</bean>

第四步

flex端代码:

<?xml version="1.0" encoding="utf-8"?>
<s:Application xmlns:fx="http://ns.adobe.com/mxml/2009"
			   xmlns:s="library://ns.adobe.com/flex/spark"
			   xmlns:mx="library://ns.adobe.com/flex/mx"
			   minWidth="955"
			   minHeight="600"
			   creationComplete="application1_creationCompleteHandler(event)">
	<fx:Script>
		<![CDATA[
			import mx.events.FlexEvent;
			import mx.messaging.ChannelSet;
			import mx.messaging.Consumer;
			import mx.messaging.channels.StreamingAMFChannel;
			import mx.messaging.events.ChannelEvent;
			import mx.messaging.events.ChannelFaultEvent;
			import mx.messaging.events.MessageEvent;
			import mx.messaging.events.MessageFaultEvent;
			import mx.rpc.remoting.RemoteObject;

			var consumer:Consumer=new Consumer();

			protected function application1_creationCompleteHandler(event:FlexEvent):void
			{
				var myStreamingAMF:StreamingAMFChannel=new StreamingAMFChannel();
				myStreamingAMF.uri="http://192.168.1.212:8080/appname/messagebroker/streamingamf";
				var channelSet:ChannelSet=new ChannelSet();
				channelSet.addChannel(myStreamingAMF);


				consumer.destination="testDataPush";
				consumer.subtopic="3d";

				consumer.channelSet=channelSet;
				consumer.addEventListener(MessageEvent.MESSAGE, messageHandler);
				consumer.addEventListener(ChannelFaultEvent.FAULT, fault);
				consumer.addEventListener(MessageFaultEvent.FAULT, fault2);

				consumer.subscribe();//订阅


			}



			private function fault(e:ChannelFaultEvent):void
			{
				trace(e.faultDetail, e.faultString);
			}

			private function fault2(e:MessageFaultEvent):void
			{
				trace(e.faultDetail, e.faultString);
			}

			private function messageHandler(event:MessageEvent):void
			{
				trace("-----messageHandler-");
				trace(event.message.body);

			}

			private function messageResult(event:MessageEvent):void
			{
				trace("-----messageResult-");

			}

			private function connectHandler(event:ChannelEvent):void
			{
				trace("----connectHandler-");

			}

			private function disconnectHandler(event:ChannelEvent):void
			{
				trace("----disconnectHandler-");

			}

			protected function button1_clickHandler(event:MouseEvent):void
			{
				var ro:RemoteObject=new RemoteObject();
				ro.endpoint="http://192.168.1.212:8080/appname/messagebroker/streamingamf";
				ro.destination="testPush";

				ro.test();
			}
		]]>
	</fx:Script>

	<s:Button label="test"
			  click="button1_clickHandler(event)"/>
</s:Application>

ok,完成

再说下整个操作流程,

当flex application构建完成时创建一个consumer对象,并向服务端发出订阅, 点击test按钮时调用服务端test 方法, 此方法中向订阅都发布消息