1 /**
2 * Licensed to the Apache Software Foundation (ASF) under one or more
3 * contributor license agreements. See the NOTICE file distributed with
4 * this work for additional information regarding copyright ownership.
5 * The ASF licenses this file to You under the Apache License, Version 2.0
6 * (the "License"); you may not use this file except in compliance with
7 * the License. You may obtain a copy of the License at
8 *
9 * http://www.apache.org/licenses/LICENSE-2.0
10 *
11 * Unless required by applicable law or agreed to in writing, software
12 * distributed under the License is distributed on an "AS IS" BASIS,
13 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 * See the License for the specific language governing permissions and
15 * limitations under the License.
16 */
17 package org.apache.openejb.server;
18
19 import org.apache.openejb.util.LogCategory;
20 import org.apache.openejb.util.Logger;
21 import org.apache.openejb.loader.Options;
22 import org.apache.openejb.loader.SystemInstance;
23 import org.apache.openejb.util.StringTemplate;
24
25 import javax.net.ServerSocketFactory;
26 import javax.net.ssl.SSLServerSocket;
27 import javax.net.ssl.SSLServerSocketFactory;
28 import java.io.IOException;
29 import java.io.InputStream;
30 import java.io.OutputStream;
31 import java.net.InetAddress;
32 import java.net.ServerSocket;
33 import java.net.Socket;
34 import java.net.SocketException;
35 import java.net.SocketTimeoutException;
36 import java.net.UnknownHostException;
37 import java.net.URI;
38 import java.util.Properties;
39 import java.util.Map;
40 import java.util.HashMap;
41 import java.util.concurrent.atomic.AtomicBoolean;
42 import java.util.concurrent.locks.Lock;
43 import java.util.concurrent.locks.ReentrantLock;
44 import java.util.concurrent.TimeUnit;
45
46 /**
47 */
48 public class ServiceDaemon implements ServerService {
49
50 private static final Logger log = Logger.getInstance(LogCategory.OPENEJB_SERVER, ServiceDaemon.class);
51
52 private ServerService next;
53
54 private SocketListener socketListener;
55
56 private int timeout;
57
58 private InetAddress address;
59
60 private int port;
61
62 private String name;
63
64 boolean stop = true;
65
66 private int backlog;
67
68 private String ip;
69
70 private boolean secure;
71 private StringTemplate discoveryUriFormat;
72 private URI uri;
73
74 public ServiceDaemon(ServerService next) {
75 this.next = next;
76 }
77
78 public ServiceDaemon(ServerService next, int port, String ip) {
79 this.port = port;
80 this.ip = ip;
81 this.address = getAddress(ip);
82 this.next = next;
83 }
84
85 public static InetAddress getAddress(String host){
86 try {
87 return InetAddress.getByName(host);
88 } catch (UnknownHostException e) {
89 throw new IllegalArgumentException(host);
90 }
91 }
92
93 public void setSoTimeout(int timeout) throws SocketException {
94 this.timeout = timeout;
95 if (socketListener != null) {
96 socketListener.setSoTimeout(timeout);
97 }
98 }
99
100 public int getSoTimeout() throws IOException {
101 if (socketListener == null) return 0;
102 return socketListener.getSoTimeout();
103 }
104
105 /**
106 * Gets the inetAddress number that the
107 * daemon is listening on.
108 */
109 public InetAddress getAddress() {
110 return address;
111 }
112
113 public void init(Properties props) throws Exception {
114
115 String formatString = props.getProperty("discovery");
116 if (formatString != null){
117 discoveryUriFormat = new StringTemplate(formatString);
118 }
119
120 ip = props.getProperty("bind");
121
122 address = getAddress(ip);
123
124 Options options = new Options(props);
125
126 port = options.get("port", 0);
127
128 int threads = options.get("threads", 100);
129
130 backlog = options.get("backlog", threads);
131
132 secure = options.get("secure", false);
133
134 timeout = 1000;
135
136 next.init(props);
137 }
138
139 public void start() throws ServiceException {
140 synchronized (this) {
141 // Don't bother if we are already started/starting
142 if (socketListener != null) {
143 return;
144 }
145
146 next.start();
147
148 ServerSocket serverSocket;
149 try {
150 if (secure) {
151 ServerSocketFactory factory = SSLServerSocketFactory.getDefault();
152 serverSocket = factory.createServerSocket(port, backlog, address);
153 final String[] enabledCipherSuites = { "SSL_DH_anon_WITH_RC4_128_MD5" };
154 ((SSLServerSocket) serverSocket).setEnabledCipherSuites(enabledCipherSuites);
155 } else {
156 serverSocket = new ServerSocket(port, backlog, address);
157 }
158
159 port = serverSocket.getLocalPort();
160 serverSocket.setSoTimeout(timeout);
161 } catch (Exception e) {
162 throw new ServiceException("Service failed to open socket", e);
163 }
164
165 socketListener = new SocketListener(next, serverSocket);
166 Thread thread = new Thread(socketListener);
167 thread.setName("service." + name + "@" + socketListener.hashCode());
168 thread.setDaemon(true);
169 thread.start();
170
171 DiscoveryAgent agent = SystemInstance.get().getComponent(DiscoveryAgent.class);
172 if (agent != null && discoveryUriFormat != null) {
173 Map<String,String> map = new HashMap<String,String>();
174 map.put("port", Integer.toString(port));
175 map.put("host", ip);
176 map.put("bind", ip);
177 String uriString = discoveryUriFormat.apply(map);
178 try {
179 uri = new URI(uriString);
180 agent.registerService(uri);
181 } catch (Exception e) {
182 log.error("Cannot register service '" + getName() + "' with DiscoveryAgent.", e);
183 }
184 }
185
186
187 }
188 }
189
190 public void stop() throws ServiceException {
191
192 synchronized (this) {
193 DiscoveryAgent agent = SystemInstance.get().getComponent(DiscoveryAgent.class);
194 if (agent != null && discoveryUriFormat != null && uri != null) {
195 try {
196 agent.unregisterService(uri);
197 } catch (IOException e) {
198 log.error("Cannot unregister service '" + getName() + "' with DiscoveryAgent.", e);
199 }
200 }
201 next.stop();
202 if (socketListener != null) {
203 socketListener.stop();
204 socketListener = null;
205 }
206 }
207 }
208
209 public String getIP() {
210 return ip;
211 }
212
213 /**
214 * Gets the port number that the
215 * daemon is listening on.
216 */
217 public int getPort() {
218 return port;
219 }
220
221 public void service(Socket socket) throws ServiceException, IOException {
222 }
223
224 public void service(InputStream in, OutputStream out) throws ServiceException, IOException {
225 }
226
227 public String getName() {
228 return next.getName();
229 }
230
231 private static class SocketListener implements Runnable {
232 private final ServerService serverService;
233 private final ServerSocket serverSocket;
234 private AtomicBoolean stop = new AtomicBoolean();
235 private Lock lock = new ReentrantLock();
236
237 public SocketListener(ServerService serverService, ServerSocket serverSocket) {
238 this.serverService = serverService;
239 this.serverSocket = serverSocket;
240 }
241
242 public void stop() {
243 stop.set(true);
244 try {
245 if (lock.tryLock(10, TimeUnit.SECONDS)){
246 serverSocket.close();
247 }
248 } catch (InterruptedException e) {
249 Thread.interrupted();
250 } catch (IOException e) {
251 }
252 }
253
254 public void run() {
255 while (!stop.get()) {
256 Socket socket = null;
257 try {
258 socket = serverSocket.accept();
259 socket.setTcpNoDelay(true);
260 if (!stop.get()) {
261 // the server service is responsible
262 // for closing the socket.
263 try {
264 lock.lock();
265 serverService.service(socket);
266 } finally {
267 lock.unlock();
268 }
269 }
270
271 // Sockets are consumed in other threads
272 // and should never be closed here
273 // It's up to the consumer of the socket
274 // to close it.
275 } catch (SocketTimeoutException e) {
276 // we don't really care
277 // log.debug("Socket timed-out",e);
278 } catch (SocketException e) {
279 if (!stop.get()){
280 log.error("Socket error", e);
281 }
282 } catch (Throwable e) {
283 log.error("Unexpected error", e);
284 }
285 }
286
287 try {
288 serverSocket.close();
289 } catch (IOException ioException) {
290 log.debug("Error cleaning up socked", ioException);
291 }
292 }
293
294 public void setSoTimeout(int timeout) throws SocketException {
295 serverSocket.setSoTimeout(timeout);
296 }
297
298 public int getSoTimeout() throws IOException {
299 return serverSocket.getSoTimeout();
300 }
301 }
302 }