import io.vertx.proton.ProtonServer; //导入依赖的package包/类
@Override
public void start(Map<String, String> props) {

    LOG.info("Start AMQP sink task");

    this.senders = new HashMap<>();

    this.serverHostname = props.get(AmqpSinkConnectorConstant.AMQP_SERVER_HOSTNAME);
    this.serverPort = Integer.valueOf(props.get(AmqpSinkConnectorConstant.AMQP_SERVER_PORT));

    Vertx vertx = Vertx.vertx();

    ProtonServer server = ProtonServer.create(vertx)
            .connectHandler((connection) -> {
                processConnection(connection);
            })
            .listen(this.serverPort, this.serverHostname, done -> {

                if (done.succeeded()) {
                    LOG.info("Listening on {}", done.result().actualPort());
                } else {
                    done.cause().printStackTrace();
                }
            });
}
 

import io.vertx.proton.ProtonServer; //导入依赖的package包/类
@Override
public void start(Map<String, String> props) {

	LOG.info("Start AMQP source task");

	this.serverHostname = props.get(AmqpSourceConnectorConstant.AMQP_SERVER_HOSTNAME);
	this.serverPort = Integer.valueOf(props.get(AmqpSourceConnectorConstant.AMQP_SERVER_PORT));
	this.receiverCredits = Integer.valueOf(props.get(AmqpSourceConnectorConstant.AMQP_SERVER_CREDITS));
	
	Vertx vertx = Vertx.vertx();
	
	ProtonServer server = ProtonServer.create(vertx)
			.connectHandler((connection) -> {
				processConnection(connection);
			})
			.listen(this.serverPort, this.serverHostname, done -> {
				
				if (done.succeeded()) {
					LOG.info("Listening on {}", done.result().actualPort());
				} else {
					done.cause().printStackTrace();
				}
			});
}
 

import io.vertx.proton.ProtonServer; //导入依赖的package包/类
@Override
public void start() {
    ProtonServerOptions options = new ProtonServerOptions();

    server = ProtonServer.create(vertx, options);
    server.saslAuthenticatorFactory(() -> saslAuthenticator);
    server.connectHandler(this::connectHandler);

    server.listen(0, "127.0.0.1", event -> {
        if(event.failed()) {
            portFuture.completeExceptionally(event.cause());
        } else {
            portFuture.complete(server.actualPort());
        }
    });
}
 

import io.vertx.proton.ProtonServer; //导入依赖的package包/类
@Override
public void start() {
    server = ProtonServer.create(vertx);
    server.connectHandler(connection -> {
        connection.openHandler(conn -> {
            log.info("Connection opened");
        }).closeHandler(conn -> {
            log.info("Connection closed");
            connection.close();
            connection.disconnect();
        }).disconnectHandler(protonConnection -> {
            log.info("Disconnected");
            connection.disconnect();
        }).sessionOpenHandler(ProtonSession::open);

        connection.senderOpenHandler(sender -> {
            senderList.add(sender);
            sendConfig(sender);
            sender.open();
        });

        connection.open();

    }).listen(listenPort);
}
 

import io.vertx.proton.ProtonServer; //导入依赖的package包/类
@Override
public void start(Future<Void> startPromise) {
    server = ProtonServer.create(vertx, options);
    server.connectHandler(this::connectHandler);
    if(authenticatorFactory != null) {
        server.saslAuthenticatorFactory(authenticatorFactory);
    }
    server.listen(port, hostname, result -> {
        if (result.succeeded()) {
            log.info("Starting server on {}:{}", hostname, port);
            startPromise.complete();
        } else {
            log.error("Error starting server", result.cause());
            startPromise.fail(result.cause());
        }
    });
}
 

import io.vertx.proton.ProtonServer; //导入依赖的package包/类
/**
 * Start the AMQP server
 *
 * @param startFuture
 */
private void bindAmqpServer(Future<Void> startFuture) {
	
	ProtonServerOptions options = this.createServerOptions();
	
	this.server = ProtonServer.create(this.vertx, options)
			.connectHandler(this::processConnection)
			.listen(ar -> {
				
				if (ar.succeeded()) {

					log.info("AMQP-Kafka Bridge started and listening on port {}", ar.result().actualPort());
					log.info("Kafka bootstrap servers {}",
							this.bridgeConfigProperties.getKafkaConfigProperties().getBootstrapServers());

					this.isReady = true;
					this.startHealthServer();

					startFuture.complete();
				} else {
					log.error("Error starting AMQP-Kafka Bridge", ar.cause());
					startFuture.fail(ar.cause());
				}
			});
}
 

import io.vertx.proton.ProtonServer; //导入依赖的package包/类
public static void main(String[] args) {

    // Create the Vert.x instance
    Vertx vertx = Vertx.vertx();

    // Create the Vert.x AMQP client instance
    ProtonServer server = ProtonServer.create(vertx).connectHandler((connection) -> {
      helloProcessConnection(vertx, connection);
    }).listen(5672, (res) -> {
      if (res.succeeded()) {
        System.out.println("Listening on: " + res.result().actualPort());
      } else {
        res.cause().printStackTrace();
      }
    });

    // Just stop main() from exiting
    try {
      System.in.read();
    } catch (Exception ignore) {
    }
  }
 

import io.vertx.proton.ProtonServer; //导入依赖的package包/类
@Test(timeout = 20000)
public void testCustomAuthenticatorSuceedsAuthentication(TestContext context) {
  Async connectedAsync = context.async();
  Async authenticatedAsync = context.async();

  ProtonServer.create(vertx).saslAuthenticatorFactory(new TestPlainAuthenticatorFactory()).connectHandler(protonConnection -> {
    // Verify the expected auth detail was recorded in the connection attachments, just using a String here.
    String authValue = protonConnection.attachments().get(AUTH_KEY, String.class);
    context.assertEquals(AUTH_VALUE, authValue);
    authenticatedAsync.complete();
  }).listen(server -> ProtonClient.create(vertx).connect("localhost", server.result().actualPort(), GOOD_USER, PASSWD,
      protonConnectionAsyncResult -> {
        context.assertTrue(protonConnectionAsyncResult.succeeded());
        protonConnectionAsyncResult.result().disconnect();
        connectedAsync.complete();
      }));

  authenticatedAsync.awaitSuccess();
  connectedAsync.awaitSuccess();
}
 

import io.vertx.proton.ProtonServer; //导入依赖的package包/类
@Override
public void start() {
    ProtonServerOptions options = new ProtonServerOptions();
    if(useTls) {
        options.setSsl(true);
        String path;
        if((path = config.get("jksKeyStorePath")) != null) {
            final JksOptions jksOptions = new JksOptions();
            jksOptions.setPath(path);
            jksOptions.setPassword(config.get("keyStorePassword"));
            options.setKeyStoreOptions(jksOptions);
        } else if((path = config.get("pfxKeyStorePath")) != null) {
            final PfxOptions pfxOptions = new PfxOptions();
            pfxOptions.setPath(path);
            pfxOptions.setPassword(config.get("keyStorePassword"));
            options.setPfxKeyCertOptions(pfxOptions);
        } else if((path = config.get("pemCertificatePath")) != null) {
            final PemKeyCertOptions pemKeyCertOptions = new PemKeyCertOptions();
            pemKeyCertOptions.setCertPath(path);
            pemKeyCertOptions.setKeyPath(config.get("pemKeyPath"));
            options.setPemKeyCertOptions(pemKeyCertOptions);
        } else {
            // use JDK settings?
        }

    }
    server = ProtonServer.create(vertx, options);

    server.saslAuthenticatorFactory(() -> new SaslAuthenticator(keycloakSessionFactory, config, useTls));
    server.connectHandler(this::connectHandler);
    LOG.info("Starting server on "+hostname+":"+ port);
    server.listen(port, hostname, event -> {
        if(event.failed())
        {
            LOG.error("Unable to listen for AMQP on "+hostname+":" + port, event.cause());
        }

    });
}
 

import io.vertx.proton.ProtonServer; //导入依赖的package包/类
private Handler<AsyncResult<NetServer>> convertHandler(final Handler<AsyncResult<ProtonServer>> handler) {
  return result -> {
    if (result.succeeded()) {
      handler.handle(Future.succeededFuture(ProtonServerImpl.this));
    } else {
      handler.handle(Future.failedFuture(result.cause()));
    }
  };
}
 

import io.vertx.proton.ProtonServer; //导入依赖的package包/类
@Override
public ProtonServer saslAuthenticatorFactory(ProtonSaslAuthenticatorFactory authenticatorFactory) {
  if (authenticatorFactory == null) {
    // restore the default
    this.authenticatorFactory = new DefaultAuthenticatorFactory();
  } else {
    this.authenticatorFactory = authenticatorFactory;
  }
  return this;
}
 

import io.vertx.proton.ProtonServer; //导入依赖的package包/类
@Test(timeout = 20000)
public void testCustomAuthenticatorHasInitCalled(TestContext context) {
  Async initCalledAsync = context.async();

  ProtonServer.create(vertx).saslAuthenticatorFactory(new ProtonSaslAuthenticatorFactory() {
    @Override
    public ProtonSaslAuthenticator create() {
      return new ProtonSaslAuthenticator() {
        @Override
        public void init(NetSocket socket, ProtonConnection protonConnection, Transport transport) {
          initCalledAsync.complete();
        }

        @Override
        public void process(Handler<Boolean> completionHandler) {
          completionHandler.handle(false);
        }

        @Override
        public boolean succeeded() {
          return false;
        }
      };
    }
  }).connectHandler(protonConnection -> {
  }).listen(server -> ProtonClient.create(vertx).connect("localhost", server.result().actualPort(),
      protonConnectionAsyncResult -> {
      }));
}
 

import io.vertx.proton.ProtonServer; //导入依赖的package包/类
@Test(timeout = 20000)
public void testCustomAuthenticatorFailsAuthentication(TestContext context) {
  Async connectedAsync = context.async();

  ProtonServer.create(vertx).saslAuthenticatorFactory(new TestPlainAuthenticatorFactory()).connectHandler(protonConnection -> {
    context.fail("Handler should not be called for connection that failed authentication");
  }).listen(server -> ProtonClient.create(vertx).connect("localhost", server.result().actualPort(), BAD_USER, PASSWD,
      protonConnectionAsyncResult -> {
        context.assertFalse(protonConnectionAsyncResult.succeeded());
        connectedAsync.complete();
      }));

  connectedAsync.awaitSuccess();
}
 

import io.vertx.proton.ProtonServer; //导入依赖的package包/类
@Test(timeout = 20000)
public void testAuthenticatorCreatedPerConnection(TestContext context) {
  Async connectedAsync = context.async();
  Async connectedAsync2 = context.async();
  AtomicInteger port = new AtomicInteger(-1);

  final TestPlainAuthenticatorFactory authenticatorFactory = new TestPlainAuthenticatorFactory();

  ProtonServer.create(vertx).saslAuthenticatorFactory(authenticatorFactory).connectHandler(protonConnection -> {
    // Verify the expected auth detail was recorded in the connection attachments, just using a String here.
    String authValue = protonConnection.attachments().get(AUTH_KEY, String.class);
    context.assertEquals(AUTH_VALUE, authValue);
  }).listen(server -> {
    port.set(server.result().actualPort());
    ProtonClient.create(vertx).connect("localhost", port.intValue(), GOOD_USER, PASSWD,
        protonConnectionAsyncResult -> {
          context.assertTrue(protonConnectionAsyncResult.succeeded());
          protonConnectionAsyncResult.result().disconnect();
          connectedAsync.complete();
        });
  });

  connectedAsync.awaitSuccess();

  context.assertEquals(1, authenticatorFactory.getCreateCount(), "unexpected authenticator count");

  ProtonClient.create(vertx).connect("localhost", port.intValue(), GOOD_USER, PASSWD, protonConnectionAsyncResult -> {
    context.assertTrue(protonConnectionAsyncResult.succeeded());
    protonConnectionAsyncResult.result().disconnect();
    connectedAsync2.complete();
  });

  connectedAsync2.awaitSuccess();

  context.assertEquals(2, authenticatorFactory.getCreateCount(), "unexpected authenticator count");
}
 

import io.vertx.proton.ProtonServer; //导入依赖的package包/类
private void doTestAsyncServerAuthenticatorTestImpl(TestContext context, boolean passAuthentication) {
  Async connectAsync = context.async();
  AtomicBoolean connectedServer = new AtomicBoolean();

  final long delay = 750;
  TestAsyncAuthenticator testAsyncAuthenticator = new TestAsyncAuthenticator(delay, passAuthentication);
  TestAsyncAuthenticatorFactory authenticatorFactory = new TestAsyncAuthenticatorFactory(testAsyncAuthenticator);

  ProtonServer.create(vertx).saslAuthenticatorFactory(authenticatorFactory).connectHandler(protonConnection -> {
    connectedServer.set(true);
  }).listen(server -> {
    final long startTime = System.currentTimeMillis();
    ProtonClient.create(vertx).connect("localhost", server.result().actualPort(), GOOD_USER, PASSWD, conResult -> {
      // Verify the process took expected time from auth delay.
      long actual = System.currentTimeMillis() - startTime;
      context.assertTrue(actual >= delay, "Connect completed before expected time delay elapsed! " + actual);

      if (passAuthentication) {
        context.assertTrue(conResult.succeeded(), "Expected connect to succeed");
        conResult.result().disconnect();
      } else {
        context.assertFalse(conResult.succeeded(), "Expected connect to fail");
      }

      connectAsync.complete();
    });
  });

  connectAsync.awaitSuccess();

  if(passAuthentication) {
    context.assertTrue(connectedServer.get(), "Server handler should have been called");
  } else {
    context.assertFalse(connectedServer.get(), "Server handler should not have been called");
  }

  context.assertEquals(1, authenticatorFactory.getCreateCount(), "unexpected authenticator creation count");
}
 

import io.vertx.proton.ProtonServer; //导入依赖的package包/类
private ProtonServer createProtonServer(final ProtonServerOptions options) {
    return ProtonServer.create(vertx, options)
            .saslAuthenticatorFactory(saslAuthenticatorFactory);
}
 

import io.vertx.proton.ProtonServer; //导入依赖的package包/类
@Override
public void start() {
    ProtonServerOptions options = new ProtonServerOptions();
    if (certDir != null) {
        options.setSsl(true)
            .setClientAuth(ClientAuth.REQUIRED)
            .setPemKeyCertOptions(new PemKeyCertOptions()
                    .setKeyPath(new File(certDir, "tls.key").getAbsolutePath())
                    .setCertPath(new File(certDir, "tls.crt").getAbsolutePath()))
            .setPemTrustOptions(new PemTrustOptions()
                    .addCertPath(new File(certDir, "ca.crt").getAbsolutePath()));
    }

    server = ProtonServer.create(vertx, options);
    server.saslAuthenticatorFactory(saslAuthenticatorFactory);
    server.connectHandler(connection -> {
        connection.setContainer("queue-scheduler");
        connection.openHandler(result -> {
            connection.open();
            connectionOpened(connection);
        }).closeHandler(conn -> {
            log.info("Broker connection " + connection.getRemoteContainer() + " closed");
            executeBlocking(() -> schedulerState.brokerRemoved(getGroupId(connection), connection.getRemoteContainer()),
                    "Error removing broker");
            connection.close();
            connection.disconnect();
        }).disconnectHandler(protonConnection -> {
            log.info("Broker connection " + connection.getRemoteContainer() + " disconnected");
            executeBlocking(() -> schedulerState.brokerRemoved(getGroupId(connection), connection.getRemoteContainer()),
                    "Error removing broker");
            connection.disconnect();
        });
    });

    server.listen(port, event -> {
        if (event.succeeded()) {
            log.info("QueueScheduler is up and running");
        } else {
            log.error("Error starting queue scheduler", event.cause());
        }
    });
}
 

import io.vertx.proton.ProtonServer; //导入依赖的package包/类
@Before
public void setup() throws InterruptedException {
    vertx = Vertx.vertx();
    server = ProtonServer.create(vertx);
    inbox = new LinkedBlockingDeque<>();
    outbox = new LinkedBlockingDeque<>();
    CountDownLatch latch = new CountDownLatch(1);
    server.connectHandler(conn -> {
        conn.closeHandler(c -> {
            conn.close();
            conn.disconnect();
        });
        conn.disconnectHandler(c -> {
            conn.disconnect();
        }).open();

        conn.sessionOpenHandler(ProtonSession::open);

        conn.receiverOpenHandler(receiver -> {
            System.out.println("Receiver open");
            receiver.setTarget(receiver.getRemoteTarget());
            receiver.handler((delivery, message) -> {
                inbox.add(message);
            });
            receiver.open();
        });

        conn.senderOpenHandler(sender -> {
            vertx.setPeriodic(100, id -> {
                try {
                    Message m = outbox.poll(0, TimeUnit.SECONDS);
                    if (m != null) {
                        sender.send(m);
                    }
                } catch (InterruptedException e) {
                    // Try again later
                }
            });
            sender.open();
        });
    }).listen(12345, res -> latch.countDown());
    latch.await();
}
 

import io.vertx.proton.ProtonServer; //导入依赖的package包/类
@Before
public void setup() throws InterruptedException {
    vertx = Vertx.vertx();
    server = ProtonServer.create(vertx);
    CountDownLatch latch = new CountDownLatch(1);
    CompletableFuture<ProtonSender> futureSender = new CompletableFuture<>();
    server.connectHandler(conn -> {
        conn.closeHandler(c -> {
            conn.close();
            conn.disconnect();
        });
        conn.disconnectHandler(c -> {
            conn.disconnect();
        }).open();

        conn.sessionOpenHandler(ProtonSession::open);

        conn.receiverOpenHandler(receiver -> {
            System.out.println("Receiver open");
            receiver.setTarget(receiver.getRemoteTarget());
            receiver.handler((delivery, message) -> {
                Message response = Message.Factory.create();
                response.setAddress(message.getAddress());
                response.setBody(new AmqpValue(true));
                response.setCorrelationId(message.getCorrelationId());
                response.setReplyTo(message.getReplyTo());
                try {
                    futureSender.get().send(response);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
            receiver.open();
        });

        conn.senderOpenHandler(sender -> {
            sender.setSource(sender.getRemoteSource());
            sender.open();
            futureSender.complete(sender);
        });
    }).listen(12347, res -> {
        latch.countDown();
    });
    latch.await();
}
 

import io.vertx.proton.ProtonServer; //导入依赖的package包/类
ProtonServer getProtonServer() {
  return server;
}