public void start() throws Exception {
        AuthenticationService authenticationService = new AuthenticationService(
                PulsarConfigurationLoader.convertFrom(config));

        if (config.getBrokerClientAuthenticationPlugin() != null) {
            proxyClientAuthentication = AuthenticationFactory.create(config.getBrokerClientAuthenticationPlugin(),
                    config.getBrokerClientAuthenticationParameters());
            Objects.requireNonNull(proxyClientAuthentication, "No supported auth found for proxy");
            try {
                proxyClientAuthentication.start();
            } catch (Exception e) {
                try {
                    proxyClientAuthentication.close();
                } catch (IOException ioe) {
                    log.error("Failed to close the authentication service", ioe);
                }
                throw new PulsarClientException.InvalidConfigurationException(e.getMessage());
            }
        } else {
            proxyClientAuthentication = AuthenticationDisabled.INSTANCE;
        }

        // create proxy service
        proxyService = new ProxyService(config, authenticationService, proxyClientAuthentication);
        // create a web-service
        server = new WebServer(config, authenticationService);

        if (!embeddedMode) {
            Runtime.getRuntime().addShutdownHook(new Thread(this::close));
        }

        proxyService.start();

        if (!metricsInitialized) {
            // Setup metrics
            DefaultExports.initialize();
            CollectorRegistry registry = CollectorRegistry.defaultRegistry;

            // Report direct memory from Netty counters
            Collector jvmMemoryDirectBytesUsed =
                    Gauge.build("jvm_memory_direct_bytes_used", "-").create().setChild(new Child() {
                        @Override
                        public double get() {
                            return getJvmDirectMemoryUsed();
                        }
                    });
            try {
                registry.register(jvmMemoryDirectBytesUsed);
            } catch (IllegalArgumentException e) {
                // workaround issue in tests where the metric is already registered
                log.debug("Failed to register jvm_memory_direct_bytes_used metric: {}", e.getMessage());
            }

            Collector jvmMemoryDirectBytesMax =
                    Gauge.build("jvm_memory_direct_bytes_max", "-").create().setChild(new Child() {
                        @Override
                        public double get() {
                            return DirectMemoryUtils.jvmMaxDirectMemory();
                        }
                    });
            try {
                registry.register(jvmMemoryDirectBytesMax);
            } catch (IllegalArgumentException e) {
                // workaround issue in tests where the metric is already registered
                log.debug("Failed to register jvm_memory_direct_bytes_max metric: {}", e.getMessage());
            }

            metricsInitialized = true;
        }

        AtomicReference<WebSocketService> webSocketServiceRef = new AtomicReference<>();
        addWebServerHandlers(server, config, proxyService, proxyService.getDiscoveryProvider(), webSocketServiceRef,
                proxyClientAuthentication);
        webSocketService = webSocketServiceRef.get();

        // start web-service
        server.start();
    }
@Test
    public void testProduceAndConsumeMessageWithWebsocket() throws Exception {
        @Cleanup("stop")
        HttpClient producerClient = new HttpClient();
        @Cleanup("stop")
        WebSocketClient producerWebSocketClient = new WebSocketClient(producerClient);
        producerWebSocketClient.start();
        MyWebSocket producerSocket = new MyWebSocket();
        String produceUri = computeWsBasePath() + "/producer/persistent/sample/test/local/websocket-topic";
        Future<Session> producerSession = producerWebSocketClient.connect(producerSocket, URI.create(produceUri));

        ProducerMessage produceRequest = new ProducerMessage();
        produceRequest.setContext("context");
        produceRequest.setPayload(Base64.getEncoder().encodeToString("my payload".getBytes()));

        @Cleanup("stop")
        HttpClient consumerClient = new HttpClient();
        @Cleanup("stop")
        WebSocketClient consumerWebSocketClient = new WebSocketClient(consumerClient);
        consumerWebSocketClient.start();
        MyWebSocket consumerSocket = new MyWebSocket();
        String consumeUri = computeWsBasePath() + "/consumer/persistent/sample/test/local/websocket-topic/my-sub";
        Future<Session> consumerSession = consumerWebSocketClient.connect(consumerSocket, URI.create(consumeUri));
        consumerSession.get().getRemote().sendPing(ByteBuffer.wrap("ping".getBytes()));
        producerSession.get().getRemote().sendString(ObjectMapperFactory.getMapper().writer().writeValueAsString(produceRequest));
        assertTrue(consumerSocket.getResponse().contains("ping"));
        ProducerMessage message = ObjectMapperFactory.getMapper().reader().readValue(consumerSocket.getResponse(), ProducerMessage.class);
        assertEquals(new String(Base64.getDecoder().decode(message.getPayload())), "my payload");
    }