001    /*
002    The contents of this file are subject to the Mozilla Public License Version 1.1 
003    (the "License"); you may not use this file except in compliance with the License. 
004    You may obtain a copy of the License at http://www.mozilla.org/MPL/ 
005    Software distributed under the License is distributed on an "AS IS" basis, 
006    WITHOUT WARRANTY OF ANY KIND, either express or implied. See the License for the 
007    specific language governing rights and limitations under the License. 
008    
009    The Original Code is "ServerSocketStreamSource.java".  Description: 
010    "A StreamSource that gets streams from ServerSockets." 
011    
012    The Initial Developer of the Original Code is University Health Network. Copyright (C) 
013    2004.  All Rights Reserved. 
014    
015    Contributor(s): ______________________________________. 
016    
017    Alternatively, the contents of this file may be used under the terms of the 
018    GNU General Public License (the  ???GPL???), in which case the provisions of the GPL are 
019    applicable instead of those above.  If you wish to allow use of your version of this 
020    file only under the terms of the GPL and not to allow others to use your version 
021    of this file under the MPL, indicate your decision by deleting  the provisions above 
022    and replace  them with the notice and other provisions required by the GPL License.  
023    If you do not delete the provisions above, a recipient may use your version of 
024    this file under either the MPL or the GPL. 
025    */
026    
027    package ca.uhn.hl7v2.protocol.impl;
028    
029    import java.io.IOException;
030    import java.net.ServerSocket;
031    import java.net.Socket;
032    
033    import ca.uhn.hl7v2.protocol.TransportException;
034    import ca.uhn.log.HapiLog;
035    import ca.uhn.log.HapiLogFactory;
036    import java.net.SocketTimeoutException;
037    
038    /**
039     * A <code>StreamSource</code> that gets streams from ServerSockets.  This 
040     * allows you to communicate over sockets that are established by the remote 
041     * party (ie as a TCP/IP server).
042     * 
043     * @author <a href="mailto:bryan.tripp@uhn.on.ca">Bryan Tripp</a>
044     * @version $Revision: 1.4 $ updated on $Date: 2009/12/19 20:01:20 $ by $Author: jamesagnew $
045     */
046    public class ServerSocketStreamSource extends SocketStreamSource {
047    
048        private ServerSocket myServerSocket;
049        private String myExpectedAddress;
050        private Socket mySocket;
051        private Acceptor myAcceptor;
052        private boolean myProceedWithConnect;
053        
054        /**
055         * @param theServerSocket a ServerSocket at which to listen for incoming connections  
056         * @param theExpectedAddress the IP address from which to accept connections (null means 
057         *      accept from any address) 
058         * @throws TransportException
059         */
060        public ServerSocketStreamSource(ServerSocket theServerSocket, String theExpectedAddress) throws TransportException {
061            myServerSocket = theServerSocket;
062            myExpectedAddress = theExpectedAddress;
063        }
064    
065        /** 
066         * @see ca.uhn.hl7v2.protocol.impl.SocketStreamSource#getSocket()
067         */
068        public Socket getSocket() {
069            return mySocket;
070        }
071    
072        /** 
073         * Accepts new connections on underlying ServerSocket, replacing 
074         * any existing socket with the new one, blocking until a connection 
075         * is available.  See {@link DualTransportConnector} for a method of 
076         * connecting two <code>TransportLayer</code>s in a way that avoids deadlock.    
077         * 
078         * @see ca.uhn.hl7v2.protocol.StreamSource#connect()
079         */
080        public void connect() throws TransportException {
081            Acceptor a = new Acceptor(myServerSocket, myExpectedAddress);                
082            mySocket = a.waitForSocket();
083        }
084        
085        /**
086         * A thing with which waiting for inbound socket connections can 
087         * be done in a separate thread.  This is needed because we may have to 
088         * start waiting at two ports before pending on either.  Otherwise if 
089         * we accept() in a different order than the remote system connects, 
090         * we will deadlock.  
091         * 
092         * @author <a href="mailto:bryan.tripp@uhn.on.ca">Bryan Tripp</a>
093         * @version $Revision: 1.4 $ updated on $Date: 2009/12/19 20:01:20 $ by $Author: jamesagnew $
094         */
095        private static class Acceptor {
096            
097            private static final HapiLog log = HapiLogFactory.getHapiLog(Acceptor.class);
098            
099            private Socket mySocket;
100            
101            /**
102             * Starts waiting in a separate thread for connections to the given 
103             * ServerSocket from the given IP address.  
104             * @param theServer 
105             * @param theAddress IP address from which to accept connections (null
106             *      means any) 
107             */
108            public Acceptor(final ServerSocket theServer, final String theAddress) {
109                final Acceptor a = this;
110                if (theAddress != null) {
111                    log.info("Server socket is about to try to accept a connection from " + theAddress);
112                } else {
113                    log.info("Server socket is about to try to accept a connection from any addess");
114                }
115    
116                Runnable r = new Runnable() {
117                    public void run() {
118                        while (true) {
119    
120                            Socket s;
121                            try {
122    
123                                    if (!theServer.isClosed()) {
124                                        s = theServer.accept();
125                                        String address = s.getInetAddress().getHostAddress();
126                                        if (theAddress == null || address.equals(theAddress)) {
127                                            a.setSocket(s);
128                                            synchronized (a) {
129                                                a.notifyAll();
130                                            }
131                                        } else {
132                                            log.info("Ignoring connection from " + address + ": expecting " + theAddress);
133                                        }
134                                    }
135                                    
136                            } catch (SocketTimeoutException e) {
137                                log.debug("Socket timed out without receiving a connection");
138                            } catch (IOException e) {
139                                log.error("Error accepting remote connection", e); 
140                            } // try-catch
141    
142                            if (a.getSocket() != null) {
143                                log.info("Accepted connection from address: " + a.getSocket().getInetAddress());
144                                return;
145                            }
146    
147                            if (theServer.isClosed()) {
148                                log.warn("Server socket closed, aborting");
149                                return;
150                            }
151    
152                            //if there's a problem, don't fill up the log at lightning speed
153                            try {
154                                Thread.sleep(1000);
155                            } catch (InterruptedException e2) {}
156    
157                        }
158                    }
159                };
160                
161                Thread thd = new Thread(r);
162                thd.start();
163            }
164            
165            public void setSocket(Socket theSocket) {
166                mySocket = theSocket;
167            }
168            
169            public Socket getSocket() {
170                return mySocket;
171            }
172            
173            /**
174             * @return as getSocket(), but doesn't return until getSocket() returns 
175             *  non-null.  
176             */
177            public Socket waitForSocket() {
178                while (getSocket() == null) {
179                    try {
180                        synchronized (this) {
181                            this.wait(100);
182                        }
183                    } catch (InterruptedException e) {}
184                }
185                return getSocket();
186            }
187            
188        }
189    
190    
191    }