Ensure DNS errors bubble up as errors in reactive streams

Currently DNS lookups happen synchronously in the reactive streams
driver.  This isn't good, but with this change at least DNS errors
will be properly reported.

JAVA-3852

Author: jyemin

driver-core/src/main/com/mongodb/internal/connection/InternalStreamConnection.java

@@ -145,37 +145,36 @@ public void openAsync(final SingleResultCallback<Void> callback) {
         isTrue("Open already called", stream == null, callback);
         try {
             stream = streamFactory.create(serverId.getAddress());
-        } catch (Throwable t) {
-            callback.onResult(null, t);
-            return;
-        }
-        stream.openAsync(new AsyncCompletionHandler<Void>() {
-            @Override
-            public void completed(final Void aVoid) {
-                connectionInitializer.initializeAsync(InternalStreamConnection.this, new SingleResultCallback<ConnectionDescription>() {
-                    @Override
-                    public void onResult(final ConnectionDescription result, final Throwable t) {
-                        if (t != null) {
-                            close();
-                            callback.onResult(null, t);
-                        } else {
-                            description = result;
-                            opened.set(true);
-                            sendCompressor = findSendCompressor(description);
-                            if (LOGGER.isInfoEnabled()) {
-                                LOGGER.info(format("Opened connection [%s] to %s", getId(), serverId.getAddress()));
+            stream.openAsync(new AsyncCompletionHandler<Void>() {
+                @Override
+                public void completed(final Void aVoid) {
+                    connectionInitializer.initializeAsync(InternalStreamConnection.this, new SingleResultCallback<ConnectionDescription>() {
+                        @Override
+                        public void onResult(final ConnectionDescription result, final Throwable t) {
+                            if (t != null) {
+                                close();
+                                callback.onResult(null, t);
+                            } else {
+                                description = result;
+                                opened.set(true);
+                                sendCompressor = findSendCompressor(description);
+                                if (LOGGER.isInfoEnabled()) {
+                                    LOGGER.info(format("Opened connection [%s] to %s", getId(), serverId.getAddress()));
+                                }
+                                callback.onResult(null, null);
                             }
-                            callback.onResult(null, null);
                         }
-                    }
-                });
-            }
+                    });
+                }
 
-            @Override
-            public void failed(final Throwable t) {
-                callback.onResult(null, t);
-            }
-        });
+                @Override
+                public void failed(final Throwable t) {
+                    callback.onResult(null, t);
+                }
+            });
+        } catch (Throwable t) {
+            callback.onResult(null, t);
+        }
     }
 
     private Map<Byte, Compressor> createCompressorMap(final List<MongoCompressor> compressorList) {