[pulsar] branch master updated: local function-worker use localhost broker service url (#2722)

[sijie]

This is an automated email from the ASF dual-hosted git repository.

sijie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git


The following commit(s) were added to refs/heads/master by this push:
     new af58956  local function-worker use localhost broker service url (#2722)
af58956 is described below

commit af589567fc5d479e2e4e205234ee2234e0eccef5
Author: Rajan Dhabalia <rdhaba...@apache.org>
AuthorDate: Fri Oct 5 00:48:31 2018 -0700

    local function-worker use localhost broker service url (#2722)
    
    ### Motivation
    
    If pulsar service starts broker and worker in the same process then worker 
should not use broker-advertised address but use localhost address to connect.
---
 .../org/apache/pulsar/PulsarBrokerStarter.java     | 13 ++++++------
 .../org/apache/pulsar/broker/PulsarService.java    | 24 ++++++++++++++++++----
 2 files changed, 27 insertions(+), 10 deletions(-)

diff --git 
a/pulsar-broker/src/main/java/org/apache/pulsar/PulsarBrokerStarter.java 
b/pulsar-broker/src/main/java/org/apache/pulsar/PulsarBrokerStarter.java
index 42e5f6e..98ed15a 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/PulsarBrokerStarter.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/PulsarBrokerStarter.java
@@ -145,12 +145,13 @@ public class PulsarBrokerStarter {
                 }
                 // worker talks to local broker
                 boolean useTls = workerConfig.isUseTls();
-                String pulsarServiceUrl = useTls && 
isNotBlank(PulsarService.brokerUrlTls(brokerConfig))
-                        ? PulsarService.brokerUrlTls(brokerConfig)
-                        : PulsarService.brokerUrl(brokerConfig);
-                String webServiceUrl = useTls && 
isNotBlank(PulsarService.webAddressTls(brokerConfig))
-                        ? PulsarService.webAddressTls(brokerConfig)
-                        : PulsarService.webAddress(brokerConfig);
+                String localhost = "127.0.0.1";
+                String pulsarServiceUrl = useTls
+                        ? PulsarService.brokerUrlTls(localhost, 
brokerConfig.getBrokerServicePortTls())
+                        : PulsarService.brokerUrl(localhost, 
brokerConfig.getBrokerServicePort());
+                String webServiceUrl = useTls
+                        ? PulsarService.webAddressTls(localhost, 
brokerConfig.getWebServicePortTls())
+                        : PulsarService.webAddress(localhost, 
brokerConfig.getWebServicePort());
                 workerConfig.setPulsarServiceUrl(pulsarServiceUrl);
                 workerConfig.setPulsarWebServiceUrl(webServiceUrl);
                 String hostname = 
ServiceConfigurationUtils.getDefaultOrConfiguredAddress(
diff --git 
a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java 
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
index ab565a5..63460c7 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
@@ -826,29 +826,45 @@ public class PulsarService implements AutoCloseable {
     }
 
     public static String brokerUrl(ServiceConfiguration config) {
-        return "pulsar://" + advertisedAddress(config) + ":" + 
config.getBrokerServicePort();
+        return brokerUrl(advertisedAddress(config), 
config.getBrokerServicePort());
+    }
+
+    public static String brokerUrl(String host, int port) {
+        return String.format("pulsar://%s:%d", host, port);
     }
 
     public static String brokerUrlTls(ServiceConfiguration config) {
         if (config.isTlsEnabled()) {
-            return "pulsar+ssl://" + advertisedAddress(config) + ":" + 
config.getBrokerServicePortTls();
+            return brokerUrlTls(advertisedAddress(config), 
config.getBrokerServicePortTls());
         } else {
             return "";
         }
     }
 
+    public static String brokerUrlTls(String host, int port) {
+        return String.format("pulsar+ssl://%s:%d", host, port);
+    }
+
     public static String webAddress(ServiceConfiguration config) {
-        return String.format("http://%s:%d";, advertisedAddress(config), 
config.getWebServicePort());
+        return webAddress(advertisedAddress(config), 
config.getWebServicePort());
+    }
+
+    public static String webAddress(String host, int port) {
+        return String.format("http://%s:%d";, host, port);
     }
 
     public static String webAddressTls(ServiceConfiguration config) {
         if (config.isTlsEnabled()) {
-            return String.format("https://%s:%d";, advertisedAddress(config), 
config.getWebServicePortTls());
+            return webAddressTls(advertisedAddress(config), 
config.getWebServicePortTls());
         } else {
             return "";
         }
     }
 
+    public static String webAddressTls(String host, int port) {
+        return String.format("https://%s:%d";, host, port);
+    }
+
     public String getBindAddress() {
         return bindAddress;
     }