1 /*
2 * Copyright (c) 1999, 2014, Oracle and/or its affiliates. All rights reserved.
3 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
4 *
5 * This code is free software; you can redistribute it and/or modify it
6 * under the terms of the GNU General Public License version 2 only, as
7 * published by the Free Software Foundation. Oracle designates this
8 * particular file as subject to the "Classpath" exception as provided
9 * by Oracle in the LICENSE file that accompanied this code.
10 *
11 * This code is distributed in the hope that it will be useful, but WITHOUT
12 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
13 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
14 * version 2 for more details (a copy is included in the LICENSE file that
15 * accompanied this code).
16 *
17 * You should have received a copy of the GNU General Public License version
18 * 2 along with this work; if not, write to the Free Software Foundation,
19 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
20 *
21 * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
22 * or visit www.oracle.com if you need additional information or have any
23 * questions.
24 */
25
26 package com.sun.jndi.ldap;
27
28 import java.io.BufferedInputStream;
29 import java.io.BufferedOutputStream;
30 import java.io.InterruptedIOException;
31 import java.io.IOException;
32 import java.io.OutputStream;
33 import java.io.InputStream;
34 import java.net.InetSocketAddress;
35 import java.net.Socket;
36 import javax.net.ssl.SSLSocket;
37
38 import javax.naming.CommunicationException;
39 import javax.naming.ServiceUnavailableException;
40 import javax.naming.NamingException;
41 import javax.naming.InterruptedNamingException;
42
43 import javax.naming.ldap.Control;
44
45 import java.lang.reflect.Method;
46 import java.lang.reflect.InvocationTargetException;
47 import java.util.Arrays;
48 import sun.misc.IOUtils;
49 import javax.net.SocketFactory;
50
51 /**
52 * A thread that creates a connection to an LDAP server.
53 * After the connection, the thread reads from the connection.
54 * A caller can invoke methods on the instance to read LDAP responses
55 * and to send LDAP requests.
56 * <p>
57 * There is a one-to-one correspondence between an LdapClient and
58 * a Connection. Access to Connection and its methods is only via
59 * LdapClient with two exceptions: SASL authentication and StartTLS.
60 * SASL needs to access Connection's socket IO streams (in order to do encryption
61 * of the security layer). StartTLS needs to do replace IO streams
62 * and close the IO streams on nonfatal close. The code for SASL
63 * authentication can be treated as being the same as from LdapClient
64 * because the SASL code is only ever called from LdapClient, from
65 * inside LdapClient's synchronized authenticate() method. StartTLS is called
66 * directly by the application but should only occur when the underlying
67 * connection is quiet.
68 * <p>
69 * In terms of synchronization, worry about data structures
70 * used by the Connection thread because that usage might contend
71 * with calls by the main threads (i.e., those that call LdapClient).
72 * Main threads need to worry about contention with each other.
73 * Fields that Connection thread uses:
74 * inStream - synced access and update; initialized in constructor;
75 * referenced outside class unsync'ed (by LdapSasl) only
76 * when connection is quiet
77 * traceFile, traceTagIn, traceTagOut - no sync; debugging only
78 * parent - no sync; initialized in constructor; no updates
79 * pendingRequests - sync
80 * pauseLock - per-instance lock;
81 * paused - sync via pauseLock (pauseReader())
82 * Members used by main threads (LdapClient):
83 * host, port - unsync; read-only access for StartTLS and debug messages
84 * setBound(), setV3() - no sync; called only by LdapClient.authenticate(),
85 * which is a sync method called only when connection is "quiet"
86 * getMsgId() - sync
87 * writeRequest(), removeRequest(),findRequest(), abandonOutstandingReqs() -
88 * access to shared pendingRequests is sync
89 * writeRequest(), abandonRequest(), ldapUnbind() - access to outStream sync
90 * cleanup() - sync
91 * readReply() - access to sock sync
92 * unpauseReader() - (indirectly via writeRequest) sync on pauseLock
93 * Members used by SASL auth (main thread):
94 * inStream, outStream - no sync; used to construct new stream; accessed
95 * only when conn is "quiet" and not shared
96 * replaceStreams() - sync method
97 * Members used by StartTLS:
98 * inStream, outStream - no sync; used to record the existing streams;
99 * accessed only when conn is "quiet" and not shared
100 * replaceStreams() - sync method
101 * <p>
102 * Handles anonymous, simple, and SASL bind for v3; anonymous and simple
103 * for v2.
104 * %%% made public for access by LdapSasl %%%
105 *
106 * @author Vincent Ryan
107 * @author Rosanna Lee
108 * @author Jagane Sundar
109 */
110 public final class Connection implements Runnable {
111
112 private static final boolean debug = false;
113 private static final int dump = 0; // > 0 r, > 1 rw
114 public static final long DEFAULT_READ_TIMEOUT_MILLIS = 15 * 1000; // 15 second timeout;
115
116
117 final private Thread worker; // Initialized in constructor
118
119 private boolean v3 = true; // Set in setV3()
120
121 final public String host; // used by LdapClient for generating exception messages
122 // used by StartTlsResponse when creating an SSL socket
123 final public int port; // used by LdapClient for generating exception messages
124 // used by StartTlsResponse when creating an SSL socket
125
126 private boolean bound = false; // Set in setBound()
127
128 // All three are initialized in constructor and read-only afterwards
129 private OutputStream traceFile = null;
130 private String traceTagIn = null;
131 private String traceTagOut = null;
132
133 // Initialized in constructor; read and used externally (LdapSasl);
134 // Updated in replaceStreams() during "quiet", unshared, period
135 public InputStream inStream; // must be public; used by LdapSasl
136
137 // Initialized in constructor; read and used externally (LdapSasl);
138 // Updated in replaceOutputStream() during "quiet", unshared, period
139 public OutputStream outStream; // must be public; used by LdapSasl
140
141 // Initialized in constructor; read and used externally (TLS) to
142 // get new IO streams; closed during cleanup
143 public Socket sock; // for TLS
144
145 // For processing "disconnect" unsolicited notification
146 // Initialized in constructor
147 final private LdapClient parent;
148
149 // Incremented and returned in sync getMsgId()
150 private int outMsgId = 0;
151
152 //
153 // The list of ldapRequests pending on this binding
154 //
155 // Accessed only within sync methods
156 private LdapRequest pendingRequests = null;
157
158 volatile IOException closureReason = null;
159 volatile boolean useable = true; // is Connection still useable
160
161 int readTimeout;
162 int connectTimeout;
163
164 // true means v3; false means v2
165 // Called in LdapClient.authenticate() (which is synchronized)
166 // when connection is "quiet" and not shared; no need to synchronize
167 void setV3(boolean v) {
168 v3 = v;
169 }
170
171 // A BIND request has been successfully made on this connection
172 // When cleaning up, remember to do an UNBIND
173 // Called in LdapClient.authenticate() (which is synchronized)
174 // when connection is "quiet" and not shared; no need to synchronize
175 void setBound() {
176 bound = true;
177 }
178
179 ////////////////////////////////////////////////////////////////////////////
180 //
181 // Create an LDAP Binding object and bind to a particular server
182 //
183 ////////////////////////////////////////////////////////////////////////////
184
185 Connection(LdapClient parent, String host, int port, String socketFactory,
186 int connectTimeout, int readTimeout, OutputStream trace) throws NamingException {
187
188 this.host = host;
189 this.port = port;
190 this.parent = parent;
191 this.readTimeout = readTimeout;
192 this.connectTimeout = connectTimeout;
193
194 if (trace != null) {
195 traceFile = trace;
196 traceTagIn = "<- " + host + ":" + port + "\n\n";
197 traceTagOut = "-> " + host + ":" + port + "\n\n";
198 }
199
200 //
201 // Connect to server
202 //
203 try {
204 sock = createSocket(host, port, socketFactory, connectTimeout);
205
206 if (debug) {
207 System.err.println("Connection: opening socket: " + host + "," + port);
208 }
209
210 inStream = new BufferedInputStream(sock.getInputStream());
211 outStream = new BufferedOutputStream(sock.getOutputStream());
212
213 } catch (InvocationTargetException e) {
214 Throwable realException = e.getTargetException();
215 // realException.printStackTrace();
216
217 CommunicationException ce =
218 new CommunicationException(host + ":" + port);
219 ce.setRootCause(realException);
220 throw ce;
221 } catch (Exception e) {
222 // We need to have a catch all here and
223 // ignore generic exceptions.
224 // Also catches all IO errors generated by socket creation.
225 CommunicationException ce =
226 new CommunicationException(host + ":" + port);
227 ce.setRootCause(e);
228 throw ce;
229 }
230
231 worker = Obj.helper.createThread(this);
232 worker.setDaemon(true);
233 worker.start();
234 }
235
236 /*
237 * Create an InetSocketAddress using the specified hostname and port number.
238 */
239 private InetSocketAddress createInetSocketAddress(String host, int port) {
240 return new InetSocketAddress(host, port);
241 }
242
243 /*
244 * Create a Socket object using the specified socket factory and time limit.
245 *
246 * If a timeout is supplied and unconnected sockets are supported then
247 * an unconnected socket is created and the timeout is applied when
248 * connecting the socket. If a timeout is supplied but unconnected sockets
249 * are not supported then the timeout is ignored and a connected socket
250 * is created.
251 */
252 private Socket createSocket(String host, int port, String socketFactory,
253 int connectTimeout) throws Exception {
254
255 Socket socket = null;
256
257 if (socketFactory != null) {
258
259 // create the factory
260
261 Class<? extends SocketFactory> socketFactoryClass =
262 (Class<? extends SocketFactory>) Obj.helper.loadClass
263 (socketFactory);
264 Method getDefault =
265 socketFactoryClass.getMethod("getDefault", new Class<?>[]{});
266 SocketFactory factory = (SocketFactory) getDefault.invoke(null, new Object[]{});
267
268 // create the socket
269
270 if (connectTimeout > 0) {
271
272 InetSocketAddress endpoint =
273 createInetSocketAddress(host, port);
274
275 // unconnected socket
276 socket = factory.createSocket();
277
278 if (debug) {
279 System.err.println("Connection: creating socket with " +
280 "a timeout using supplied socket factory");
281 }
282
283 // connected socket
284 socket.connect(endpoint, connectTimeout);
285 }
286
287 // continue (but ignore connectTimeout)
288 if (socket == null) {
289 if (debug) {
290 System.err.println("Connection: creating socket using " +
291 "supplied socket factory");
292 }
293 // connected socket
294 socket = factory.createSocket(host, port);
295 }
296 } else {
297
298 if (connectTimeout > 0) {
299
300 InetSocketAddress endpoint = createInetSocketAddress(host, port);
301
302 socket = new Socket();
303
304 if (debug) {
305 System.err.println("Connection: creating socket with " +
306 "a timeout");
307 }
308 socket.connect(endpoint, connectTimeout);
309 }
310
311 // continue (but ignore connectTimeout)
312
313 if (socket == null) {
314 if (debug) {
315 System.err.println("Connection: creating socket");
316 }
317 // connected socket
318 socket = new Socket(host, port);
319 }
320 }
321
322 // For LDAP connect timeouts on LDAP over SSL connections must treat
323 // the SSL handshake following socket connection as part of the timeout.
324 // So explicitly set a socket read timeout, trigger the SSL handshake,
325 // then reset the timeout.
326 if (connectTimeout > 0 && socket instanceof SSLSocket) {
327 SSLSocket sslSocket = (SSLSocket) socket;
328 int socketTimeout = sslSocket.getSoTimeout();
329
330 sslSocket.setSoTimeout(connectTimeout); // reuse full timeout value
331 sslSocket.startHandshake();
332 sslSocket.setSoTimeout(socketTimeout);
333 }
334
335 return socket;
336 }
337
338 ////////////////////////////////////////////////////////////////////////////
339 //
340 // Methods to IO to the LDAP server
341 //
342 ////////////////////////////////////////////////////////////////////////////
343
344 synchronized int getMsgId() {
345 return ++outMsgId;
346 }
347
348 LdapRequest writeRequest(BerEncoder ber, int msgId) throws IOException {
349 return writeRequest(ber, msgId, false /* pauseAfterReceipt */, -1);
350 }
351
352 LdapRequest writeRequest(BerEncoder ber, int msgId,
353 boolean pauseAfterReceipt) throws IOException {
354 return writeRequest(ber, msgId, pauseAfterReceipt, -1);
355 }
356
357 LdapRequest writeRequest(BerEncoder ber, int msgId,
358 boolean pauseAfterReceipt, int replyQueueCapacity) throws IOException {
359
360 LdapRequest req =
361 new LdapRequest(msgId, pauseAfterReceipt, replyQueueCapacity);
362 addRequest(req);
363
364 if (traceFile != null) {
365 Ber.dumpBER(traceFile, traceTagOut, ber.getBuf(), 0, ber.getDataLen());
366 }
367
368
369 // unpause reader so that it can get response
370 // NOTE: Must do this before writing request, otherwise might
371 // create a race condition where the writer unblocks its own response
372 unpauseReader();
373
374 if (debug) {
375 System.err.println("Writing request to: " + outStream);
376 }
377
378 try {
379 synchronized (this) {
380 outStream.write(ber.getBuf(), 0, ber.getDataLen());
381 outStream.flush();
382 }
383 } catch (IOException e) {
384 cleanup(null, true);
385 throw (closureReason = e); // rethrow
386 }
387
388 return req;
389 }
390
391 /**
392 * Reads a reply; waits until one is ready.
393 */
394 BerDecoder readReply(LdapRequest ldr)
395 throws IOException, NamingException {
396 BerDecoder rber;
397 boolean waited = false;
398
399 while (((rber = ldr.getReplyBer()) == null) && !waited) {
400 try {
401 // If socket closed, don't even try
402 synchronized (this) {
403 if (sock == null) {
404 throw new ServiceUnavailableException(host + ":" + port +
405 "; socket closed");
406 }
407 }
408 synchronized (ldr) {
409 // check if condition has changed since our last check
410 rber = ldr.getReplyBer();
411 if (rber == null) {
412 if (readTimeout > 0) { // Socket read timeout is specified
413
414 // will be woken up before readTimeout only if reply is
415 // available
416 ldr.wait(readTimeout);
417 } else {
418 ldr.wait(DEFAULT_READ_TIMEOUT_MILLIS);
419 }
420 waited = true;
421 } else {
422 break;
423 }
424 }
425 } catch (InterruptedException ex) {
426 throw new InterruptedNamingException(
427 "Interrupted during LDAP operation");
428 }
429 }
430
431 if ((rber == null) && waited) {
432 abandonRequest(ldr, null);
433 throw new NamingException("LDAP response read timed out, timeout used:"
434 + readTimeout + "ms." );
435
436 }
437 return rber;
438 }
439
440
441 ////////////////////////////////////////////////////////////////////////////
442 //
443 // Methods to add, find, delete, and abandon requests made to server
444 //
445 ////////////////////////////////////////////////////////////////////////////
446
447 private synchronized void addRequest(LdapRequest ldapRequest) {
448
449 LdapRequest ldr = pendingRequests;
450 if (ldr == null) {
451 pendingRequests = ldapRequest;
452 ldapRequest.next = null;
453 } else {
454 ldapRequest.next = pendingRequests;
455 pendingRequests = ldapRequest;
456 }
457 }
458
459 synchronized LdapRequest findRequest(int msgId) {
460
461 LdapRequest ldr = pendingRequests;
462 while (ldr != null) {
463 if (ldr.msgId == msgId) {
464 return ldr;
465 }
466 ldr = ldr.next;
467 }
468 return null;
469
470 }
471
472 synchronized void removeRequest(LdapRequest req) {
473 LdapRequest ldr = pendingRequests;
474 LdapRequest ldrprev = null;
475
476 while (ldr != null) {
477 if (ldr == req) {
478 ldr.cancel();
479
480 if (ldrprev != null) {
481 ldrprev.next = ldr.next;
482 } else {
483 pendingRequests = ldr.next;
484 }
485 ldr.next = null;
486 }
487 ldrprev = ldr;
488 ldr = ldr.next;
489 }
490 }
491
492 void abandonRequest(LdapRequest ldr, Control[] reqCtls) {
493 // Remove from queue
494 removeRequest(ldr);
495
496 BerEncoder ber = new BerEncoder(256);
497 int abandonMsgId = getMsgId();
498
499 //
500 // build the abandon request.
501 //
502 try {
503 ber.beginSeq(Ber.ASN_SEQUENCE | Ber.ASN_CONSTRUCTOR);
504 ber.encodeInt(abandonMsgId);
505 ber.encodeInt(ldr.msgId, LdapClient.LDAP_REQ_ABANDON);
506
507 if (v3) {
508 LdapClient.encodeControls(ber, reqCtls);
509 }
510 ber.endSeq();
511
512 if (traceFile != null) {
513 Ber.dumpBER(traceFile, traceTagOut, ber.getBuf(), 0,
514 ber.getDataLen());
515 }
516
517 synchronized (this) {
518 outStream.write(ber.getBuf(), 0, ber.getDataLen());
519 outStream.flush();
520 }
521
522 } catch (IOException ex) {
523 //System.err.println("ldap.abandon: " + ex);
524 }
525
526 // Don't expect any response for the abandon request.
527 }
528
529 synchronized void abandonOutstandingReqs(Control[] reqCtls) {
530 LdapRequest ldr = pendingRequests;
531
532 while (ldr != null) {
533 abandonRequest(ldr, reqCtls);
534 pendingRequests = ldr = ldr.next;
535 }
536 }
537
538 ////////////////////////////////////////////////////////////////////////////
539 //
540 // Methods to unbind from server and clear up resources when object is
541 // destroyed.
542 //
543 ////////////////////////////////////////////////////////////////////////////
544
545 private void ldapUnbind(Control[] reqCtls) {
546
547 BerEncoder ber = new BerEncoder(256);
548 int unbindMsgId = getMsgId();
549
550 //
551 // build the unbind request.
552 //
553
554 try {
555
556 ber.beginSeq(Ber.ASN_SEQUENCE | Ber.ASN_CONSTRUCTOR);
557 ber.encodeInt(unbindMsgId);
558 // IMPLICIT TAGS
559 ber.encodeByte(LdapClient.LDAP_REQ_UNBIND);
560 ber.encodeByte(0);
561
562 if (v3) {
563 LdapClient.encodeControls(ber, reqCtls);
564 }
565 ber.endSeq();
566
567 if (traceFile != null) {
568 Ber.dumpBER(traceFile, traceTagOut, ber.getBuf(),
569 0, ber.getDataLen());
570 }
571
572 synchronized (this) {
573 outStream.write(ber.getBuf(), 0, ber.getDataLen());
574 outStream.flush();
575 }
576
577 } catch (IOException ex) {
578 //System.err.println("ldap.unbind: " + ex);
579 }
580
581 // Don't expect any response for the unbind request.
582 }
583
584 /**
585 * @param reqCtls Possibly null request controls that accompanies the
586 * abandon and unbind LDAP request.
587 * @param notifyParent true means to call parent LdapClient back, notifying
588 * it that the connection has been closed; false means not to notify
589 * parent. If LdapClient invokes cleanup(), notifyParent should be set to
590 * false because LdapClient already knows that it is closing
591 * the connection. If Connection invokes cleanup(), notifyParent should be
592 * set to true because LdapClient needs to know about the closure.
593 */
594 void cleanup(Control[] reqCtls, boolean notifyParent) {
595 boolean nparent = false;
596
597 synchronized (this) {
598 useable = false;
599
600 if (sock != null) {
601 if (debug) {
602 System.err.println("Connection: closing socket: " + host + "," + port);
603 }
604 try {
605 if (!notifyParent) {
606 abandonOutstandingReqs(reqCtls);
607 }
608 if (bound) {
609 ldapUnbind(reqCtls);
610 }
611 } finally {
612 try {
613 outStream.flush();
614 sock.close();
615 unpauseReader();
616 } catch (IOException ie) {
617 if (debug)
618 System.err.println("Connection: problem closing socket: " + ie);
619 }
620 if (!notifyParent) {
621 LdapRequest ldr = pendingRequests;
622 while (ldr != null) {
623 ldr.cancel();
624 ldr = ldr.next;
625 }
626 }
627 sock = null;
628 }
629 nparent = notifyParent;
630 }
631 if (nparent) {
632 LdapRequest ldr = pendingRequests;
633 while (ldr != null) {
634
635 synchronized (ldr) {
636 ldr.notify();
637 ldr = ldr.next;
638 }
639 }
640 }
641 }
642 if (nparent) {
643 parent.processConnectionClosure();
644 }
645 }
646
647
648 // Assume everything is "quiet"
649 // "synchronize" might lead to deadlock so don't synchronize method
650 // Use streamLock instead for synchronizing update to stream
651
652 synchronized public void replaceStreams(InputStream newIn, OutputStream newOut) {
653 if (debug) {
654 System.err.println("Replacing " + inStream + " with: " + newIn);
655 System.err.println("Replacing " + outStream + " with: " + newOut);
656 }
657
658 inStream = newIn;
659
660 // Cleanup old stream
661 try {
662 outStream.flush();
663 } catch (IOException ie) {
664 if (debug)
665 System.err.println("Connection: cannot flush outstream: " + ie);
666 }
667
668 // Replace stream
669 outStream = newOut;
670 }
671
672 /**
673 * Used by Connection thread to read inStream into a local variable.
674 * This ensures that there is no contention between the main thread
675 * and the Connection thread when the main thread updates inStream.
676 */
677 synchronized private InputStream getInputStream() {
678 return inStream;
679 }
680
681
682 ////////////////////////////////////////////////////////////////////////////
683 //
684 // Code for pausing/unpausing the reader thread ('worker')
685 //
686 ////////////////////////////////////////////////////////////////////////////
687
688 /*
689 * The main idea is to mark requests that need the reader thread to
690 * pause after getting the response. When the reader thread gets the response,
691 * it waits on a lock instead of returning to the read(). The next time a
692 * request is sent, the reader is automatically unblocked if necessary.
693 * Note that the reader must be unblocked BEFORE the request is sent.
694 * Otherwise, there is a race condition where the request is sent and
695 * the reader thread might read the response and be unblocked
696 * by writeRequest().
697 *
698 * This pause gives the main thread (StartTLS or SASL) an opportunity to
699 * update the reader's state (e.g., its streams) if necessary.
700 * The assumption is that the connection will remain quiet during this pause
701 * (i.e., no intervening requests being sent).
702 *<p>
703 * For dealing with StartTLS close,
704 * when the read() exits either due to EOF or an exception,
705 * the reader thread checks whether there is a new stream to read from.
706 * If so, then it reattempts the read. Otherwise, the EOF or exception
707 * is processed and the reader thread terminates.
708 * In a StartTLS close, the client first replaces the SSL IO streams with
709 * plain ones and then closes the SSL socket.
710 * If the reader thread attempts to read, or was reading, from
711 * the SSL socket (that is, it got to the read BEFORE replaceStreams()),
712 * the SSL socket close will cause the reader thread to
713 * get an EOF/exception and reexamine the input stream.
714 * If the reader thread sees a new stream, it reattempts the read.
715 * If the underlying socket is still alive, then the new read will succeed.
716 * If the underlying socket has been closed also, then the new read will
717 * fail and the reader thread exits.
718 * If the reader thread attempts to read, or was reading, from the plain
719 * socket (that is, it got to the read AFTER replaceStreams()), the
720 * SSL socket close will have no effect on the reader thread.
721 *
722 * The check for new stream is made only
723 * in the first attempt at reading a BER buffer; the reader should
724 * never be in midst of reading a buffer when a nonfatal close occurs.
725 * If this occurs, then the connection is in an inconsistent state and
726 * the safest thing to do is to shut it down.
727 */
728
729 private Object pauseLock = new Object(); // lock for reader to wait on while paused
730 private boolean paused = false; // paused state of reader
731
732 /*
733 * Unpauses reader thread if it was paused
734 */
735 private void unpauseReader() throws IOException {
736 synchronized (pauseLock) {
737 if (paused) {
738 if (debug) {
739 System.err.println("Unpausing reader; read from: " +
740 inStream);
741 }
742 paused = false;
743 pauseLock.notify();
744 }
745 }
746 }
747
748 /*
749 * Pauses reader so that it stops reading from the input stream.
750 * Reader blocks on pauseLock instead of read().
751 * MUST be called from within synchronized (pauseLock) clause.
752 */
753 private void pauseReader() throws IOException {
754 if (debug) {
755 System.err.println("Pausing reader; was reading from: " +
756 inStream);
757 }
758 paused = true;
759 try {
760 while (paused) {
761 pauseLock.wait(); // notified by unpauseReader
762 }
763 } catch (InterruptedException e) {
764 throw new InterruptedIOException(
765 "Pause/unpause reader has problems.");
766 }
767 }
768
769
770 ////////////////////////////////////////////////////////////////////////////
771 //
772 // The LDAP Binding thread. It does the mux/demux of multiple requests
773 // on the same TCP connection.
774 //
775 ////////////////////////////////////////////////////////////////////////////
776
777
778 public void run() {
779 byte inbuf[]; // Buffer for reading incoming bytes
780 int inMsgId; // Message id of incoming response
781 int bytesread; // Number of bytes in inbuf
782 int br; // Temp; number of bytes read from stream
783 int offset; // Offset of where to store bytes in inbuf
784 int seqlen; // Length of ASN sequence
785 int seqlenlen; // Number of sequence length bytes
786 boolean eos; // End of stream
787 BerDecoder retBer; // Decoder for ASN.1 BER data from inbuf
788 InputStream in = null;
789
790 try {
791 while (true) {
792 try {
793 // type and length (at most 128 octets for long form)
794 inbuf = new byte[129];
795
796 offset = 0;
797 seqlen = 0;
798 seqlenlen = 0;
799
800 in = getInputStream();
801
802 // check that it is the beginning of a sequence
803 bytesread = in.read(inbuf, offset, 1);
804 if (bytesread < 0) {
805 if (in != getInputStream()) {
806 continue; // a new stream to try
807 } else {
808 break; // EOF
809 }
810 }
811
812 if (inbuf[offset++] != (Ber.ASN_SEQUENCE | Ber.ASN_CONSTRUCTOR))
813 continue;
814
815 // get length of sequence
816 bytesread = in.read(inbuf, offset, 1);
817 if (bytesread < 0)
818 break; // EOF
819 seqlen = inbuf[offset++];
820
821 // if high bit is on, length is encoded in the
822 // subsequent length bytes and the number of length bytes
823 // is equal to & 0x80 (i.e. length byte with high bit off).
824 if ((seqlen & 0x80) == 0x80) {
825 seqlenlen = seqlen & 0x7f; // number of length bytes
826
827 bytesread = 0;
828 eos = false;
829
830 // Read all length bytes
831 while (bytesread < seqlenlen) {
832 br = in.read(inbuf, offset+bytesread,
833 seqlenlen-bytesread);
834 if (br < 0) {
835 eos = true;
836 break; // EOF
837 }
838 bytesread += br;
839 }
840
841 // end-of-stream reached before length bytes are read
842 if (eos)
843 break; // EOF
844
845 // Add contents of length bytes to determine length
846 seqlen = 0;
847 for( int i = 0; i < seqlenlen; i++) {
848 seqlen = (seqlen << 8) + (inbuf[offset+i] & 0xff);
849 }
850 offset += bytesread;
851 }
852
853 // read in seqlen bytes
854 byte[] left = IOUtils.readFully(in, seqlen, false);
855 inbuf = Arrays.copyOf(inbuf, offset + left.length);
856 System.arraycopy(left, 0, inbuf, offset, left.length);
857 offset += left.length;
858 /*
859 if (dump > 0) {
860 System.err.println("seqlen: " + seqlen);
861 System.err.println("bufsize: " + offset);
862 System.err.println("bytesleft: " + bytesleft);
863 System.err.println("bytesread: " + bytesread);
864 }
865 */
866
867
868 try {
869 retBer = new BerDecoder(inbuf, 0, offset);
870
871 if (traceFile != null) {
872 Ber.dumpBER(traceFile, traceTagIn, inbuf, 0, offset);
873 }
874
875 retBer.parseSeq(null);
876 inMsgId = retBer.parseInt();
877 retBer.reset(); // reset offset
878
879 boolean needPause = false;
880
881 if (inMsgId == 0) {
882 // Unsolicited Notification
883 parent.processUnsolicited(retBer);
884 } else {
885 LdapRequest ldr = findRequest(inMsgId);
886
887 if (ldr != null) {
888
889 /**
890 * Grab pauseLock before making reply available
891 * to ensure that reader goes into paused state
892 * before writer can attempt to unpause reader
893 */
894 synchronized (pauseLock) {
895 needPause = ldr.addReplyBer(retBer);
896 if (needPause) {
897 /*
898 * Go into paused state; release
899 * pauseLock
900 */
901 pauseReader();
902 }
903
904 // else release pauseLock
905 }
906 } else {
907 // System.err.println("Cannot find" +
908 // "LdapRequest for " + inMsgId);
909 }
910 }
911 } catch (Ber.DecodeException e) {
912 //System.err.println("Cannot parse Ber");
913 }
914 } catch (IOException ie) {
915 if (debug) {
916 System.err.println("Connection: Inside Caught " + ie);
917 ie.printStackTrace();
918 }
919
920 if (in != getInputStream()) {
921 // A new stream to try
922 // Go to top of loop and continue
923 } else {
924 if (debug) {
925 System.err.println("Connection: rethrowing " + ie);
926 }
927 throw ie; // rethrow exception
928 }
929 }
930 }
931
932 if (debug) {
933 System.err.println("Connection: end-of-stream detected: "
934 + in);
935 }
936 } catch (IOException ex) {
937 if (debug) {
938 System.err.println("Connection: Caught " + ex);
939 }
940 closureReason = ex;
941 } finally {
942 cleanup(null, true); // cleanup
943 }
944 if (debug) {
945 System.err.println("Connection: Thread Exiting");
946 }
947 }
948
949
950 // This code must be uncommented to run the LdapAbandonTest.
951 /*public void sendSearchReqs(String dn, int numReqs) {
952 int i;
953 String attrs[] = null;
954 for(i = 1; i <= numReqs; i++) {
955 BerEncoder ber = new BerEncoder(2048);
956
957 try {
958 ber.beginSeq(Ber.ASN_SEQUENCE | Ber.ASN_CONSTRUCTOR);
959 ber.encodeInt(i);
960 ber.beginSeq(LdapClient.LDAP_REQ_SEARCH);
961 ber.encodeString(dn == null ? "" : dn);
962 ber.encodeInt(0, LdapClient.LBER_ENUMERATED);
963 ber.encodeInt(3, LdapClient.LBER_ENUMERATED);
964 ber.encodeInt(0);
965 ber.encodeInt(0);
966 ber.encodeBoolean(true);
967 LdapClient.encodeFilter(ber, "");
968 ber.beginSeq(Ber.ASN_SEQUENCE | Ber.ASN_CONSTRUCTOR);
969 ber.encodeStringArray(attrs);
970 ber.endSeq();
971 ber.endSeq();
972 ber.endSeq();
973 writeRequest(ber, i);
974 //System.err.println("wrote request " + i);
975 } catch (Exception ex) {
976 //System.err.println("ldap.search: Caught " + ex + " building req");
977 }
978
979 }
980 } */
981 }