1
22
23 package com.liferay.portal.comm;
24
25 import com.liferay.portal.kernel.util.MethodInvoker;
26 import com.liferay.portal.kernel.util.MethodWrapper;
27 import com.liferay.portal.kernel.util.Validator;
28 import com.liferay.portal.util.PropsKeys;
29 import com.liferay.portal.util.PropsUtil;
30
31 import java.io.Serializable;
32
33 import java.util.LinkedList;
34 import java.util.List;
35
36 import org.apache.commons.logging.Log;
37 import org.apache.commons.logging.LogFactory;
38
39 import org.jgroups.Channel;
40 import org.jgroups.JChannel;
41 import org.jgroups.Message;
42 import org.jgroups.MessageListener;
43 import org.jgroups.blocks.PullPushAdapter;
44 import org.jgroups.util.Util;
45
46
52 public class CommLink implements MessageListener {
53
54 public static CommLink getInstance() {
55 return _instance;
56 }
57
58 public byte[] getState() {
59 try {
60 return Util.objectToByteBuffer(_history);
61 }
62 catch (Exception e) {
63 return null;
64 }
65 }
66
67 public void setState(byte[] state) {
68 try {
69 _history = (List<Object>)Util.objectFromByteBuffer(state);
70 }
71 catch (Exception e) {
72 _log.error("Error setting state", e);
73 }
74 }
75
76 public void receive(Message msg) {
77 try {
78 Object obj = msg.getObject();
79
80 if (_log.isDebugEnabled()) {
81 _log.debug("Receiving message " + obj);
82 }
83
84 if (obj instanceof MethodWrapper) {
85 MethodWrapper methodWrapper = (MethodWrapper)obj;
86
87 MethodInvoker.invoke(methodWrapper);
88 }
89
90 _history.add(obj);
91 }
92 catch (Exception e) {
93 _log.error("Error receiving message", e);
94 }
95 }
96
97 public void send(Serializable obj) {
98 try {
99 if (_channel == null) {
100 return;
101 }
102
103 Message msg = new Message(null, null, obj);
104
105 _channel.send(msg);
106 }
107 catch (Exception e) {
108 _log.error("Error sending message", e);
109 }
110 }
111
112 private CommLink() {
113 try {
114 String properties = PropsUtil.get(PropsKeys.COMM_LINK_PROPERTIES);
115
116 if (Validator.isNotNull(properties)) {
117 _channel = new JChannel(properties);
118
119 _channel.connect("PortalMessageListener");
120
121 new PullPushAdapter(_channel, this);
122 }
123 }
124 catch (Exception e) {
125 _log.error("Error initializing", e);
126 }
127 }
128
129 private static Log _log = LogFactory.getLog(CommLink.class);
130
131 private static CommLink _instance = new CommLink();
132
133 private Channel _channel;
134 private List<Object> _history = new LinkedList<Object>();
135
136 }