1
14
15 package com.liferay.util.transport;
16
17 import java.io.IOException;
18
19 import java.net.DatagramPacket;
20 import java.net.InetAddress;
21 import java.net.MulticastSocket;
22
23 import org.apache.commons.logging.Log;
24 import org.apache.commons.logging.LogFactory;
25
26
37 public class MulticastTransport extends Thread implements Transport {
38
39 public MulticastTransport(DatagramHandler handler, String host, int port) {
40 super("MulticastListener-" + host + port);
41
42 setDaemon(true);
43 _handler = handler;
44 _host = host;
45 _port = port;
46 }
47
48 public synchronized void connect() throws IOException {
49 if (_socket == null) {
50 _socket = new MulticastSocket(_port);
51 }
52 else if (_socket.isConnected() && _socket.isBound()) {
53 return;
54 }
55
56 _address = InetAddress.getByName(_host);
57
58 _socket.joinGroup(_address);
59
60 _connected = true;
61
62 start();
63 }
64
65 public synchronized void disconnect() {
66
67
69 if (_address != null) {
70 try {
71 _socket.leaveGroup(_address);
72 _address = null;
73 }
74 catch (IOException e) {
75 _log.error("Unable to leave group", e);
76 }
77 }
78
79 _connected = false;
80
81 interrupt();
82
83 _socket.close();
84 }
85
86 public synchronized void sendMessage(String msg) throws IOException {
87 _outboundPacket.setData(msg.getBytes());
88 _outboundPacket.setAddress(_address);
89 _outboundPacket.setPort(_port);
90
91 _socket.send(_outboundPacket);
92 }
93
94 public boolean isConnected() {
95 return _connected;
96 }
97
98 public void run() {
99 try {
100 while (_connected) {
101 _socket.receive(_inboundPacket);
102 _handler.process(_inboundPacket);
103 }
104 }
105 catch (IOException e) {
106 _log.error("Unable to process ", e);
107
108 _socket.disconnect();
109
110 _connected = false;
111
112 _handler.errorReceived(e);
113 }
114 }
115
116 private static Log _log = LogFactory.getLog(MulticastTransport.class);
117
118 private byte[] _inboundBuffer = new byte[4096];
119 private DatagramPacket _inboundPacket =
120 new DatagramPacket(_inboundBuffer, _inboundBuffer.length);
121 private byte[] _outboundBuffer = new byte[4096];
122 private DatagramPacket _outboundPacket =
123 new DatagramPacket(_outboundBuffer, _outboundBuffer.length);
124 private String _host;
125 private DatagramHandler _handler;
126 private int _port;
127 private boolean _connected;
128 private MulticastSocket _socket;
129 private InetAddress _address;
130
131 }