import ws.wamp.jawampa.WampClient; //导入依赖的package包/类
public void init() throws Exception {
  WampClientBuilder builder = new WampClientBuilder();
  builder.withConnectorProvider(new NettyWampClientConnectorProvider())
      .withAuthId(config.wampUsername())
      .withAuthMethod(new Ticket(config.wampPassword()))
      .withUri(config.wampUri())
      .withRealm(config.wampRealm())
      .withInfiniteReconnects()
      .withReconnectInterval(config.wampReconnectInterval(),
          TimeUnit.SECONDS);
  client = builder.build();
  client.open();
  client.statusChanged().subscribe((WampClient.State newStatus) -> {
    if (newStatus instanceof WampClient.ConnectedState) {
      logger.info("Connected to {}",
          config.wampUri());
    } else if (newStatus instanceof WampClient.DisconnectedState) {
      logger.info("Disconnected from {}",
          config.wampUri());
    } else if (newStatus instanceof WampClient.ConnectingState) {
      logger.debug("Connecting to {}",
          config.wampUri());
    }
  });
}
 

import ws.wamp.jawampa.WampClient; //导入依赖的package包/类
public Observable<WampClient.State> init(
    String wampUri, String wampRealm, int wampReconnectInterval,
    String authId, String ticket)
    throws Exception {
  WampClientBuilder builder = new WampClientBuilder();
  builder.withConnectorProvider(new PlainWampClientConnectorProvider())
      .withUri(wampUri)
      .withRealm(wampRealm)
      .withInfiniteReconnects()
      .withReconnectInterval(wampReconnectInterval,
          TimeUnit.SECONDS)
      .withAuthId(authId)
      .withAuthMethod(new Ticket(ticket));
  client = builder.build();
  client.open();
  return client.statusChanged();
}
 

import ws.wamp.jawampa.WampClient; //导入依赖的package包/类
@Override
public void init() {
    try {
    WampClientBuilder builder = new WampClientBuilder();
        builder.witUri(serverURI)
                .withRealm(realm)
                .withSslContext(SslContext.newClientContext(
                        InsecureTrustManagerFactory.INSTANCE))
                .withInfiniteReconnects()
                .withReconnectInterval(5, TimeUnit.SECONDS);
        client = builder.build();
        client.statusChanged().subscribe((WampClient.Status newStatus) -> {
            if(WampClient.Status.Connected == newStatus) {
                logger.debug("Connected to WAMP router [{}]", serverURI);
            } else if(WampClient.Status.Connecting == newStatus) {
                logger.debug("Connecting to WAMP router [{}]", serverURI);
            }
        });
        client.open();
    } catch(Exception ex) {
        logger.error(ex.getMessage(), ex);
    }
}
 

import ws.wamp.jawampa.WampClient; //导入依赖的package包/类
private CompletableFuture<WampClient> getOrCreateClient(final StreamURI streamURI)
        throws ApplicationError {
    CompletableFuture<WampClient> result = new CompletableFuture<>();
    if (!clients.containsKey(streamURI.getServerURI())) {
        WampClientBuilder builder = new WampClientBuilder();
        builder.witUri(streamURI.getServerURI().toASCIIString())
                .withRealm(DEFAULT_REALM)
                .withInfiniteReconnects()
                .withReconnectInterval(5, TimeUnit.SECONDS);
        WampClient client = builder.build();
        client.statusChanged().subscribe((WampClient.Status newStatus) -> {
            logger.debug("WAMP router [{}] status: {}", 
                    streamURI.getServerURI(), newStatus);
            if(newStatus == WampClient.Status.Connected) {
                result.complete(client);
            }
        });
        client.open();
        clients.put(streamURI.getServerURI(), client);
    } else {
        result.complete(clients.get(streamURI.getServerURI()));
    }
    return result;
}
 

import ws.wamp.jawampa.WampClient; //导入依赖的package包/类
void closeSession(Throwable disconnectReason, String optCloseMessageReason, boolean reconnectAllowed) {
    // Send goodbye message with close reason to the remote
    if (optCloseMessageReason != null) {
        GoodbyeMessage msg = new GoodbyeMessage(null, optCloseMessageReason);
        connectionController.sendMessage(msg, IWampConnectionPromise.Empty);
    }
    
    stateController.setExternalState(new WampClient.DisconnectedState(disconnectReason));
    
    int nrReconnectAttempts = reconnectAllowed ? stateController.clientConfig().totalNrReconnects() : 0;
    if (nrReconnectAttempts != 0) {
        stateController.setExternalState(new WampClient.ConnectingState());
    }
    
    clearSessionData();
    
    WaitingForDisconnectState newState = new WaitingForDisconnectState(stateController, nrReconnectAttempts);
    connectionController.close(true, newState.closePromise());
    stateController.setState(newState);
}
 

import ws.wamp.jawampa.WampClient; //导入依赖的package包/类
public Observable<WampClient.State> init() throws Exception {
  WampClientBuilder builder = new WampClientBuilder();
  builder.withUri(CONFIG.wampUri()).withRealm(CONFIG.wampRealm())
      .withInfiniteReconnects()
      .withReconnectInterval(CONFIG.wampReconnectInterval(), TimeUnit.SECONDS)
      .withConnectorProvider(new PlainWampClientConnectorProvider())
      .withAuthId(CONFIG.wampLogin())
      .withAuthMethod(new Ticket(CONFIG.wampPassword()));
  client = builder.build();
  client.open();
  return client.statusChanged();
}
 

import ws.wamp.jawampa.WampClient; //导入依赖的package包/类
@PostConstruct
public void init() {
  logger.info("initializing");
  final IWampConnectorProvider provider = new NettyWampClientConnectorProvider();

  try {
    WampClientBuilder builder = new WampClientBuilder();
    builder
        .withConnectorProvider(provider)
        .withUri(config.wampUri())
        .withRealm(config.wampRealm())
        .withInfiniteReconnects()
        .withReconnectInterval(5, TimeUnit.SECONDS);
    client = builder.build();

    client.statusChanged().subscribe((WampClient.State state) -> {
      if (state instanceof WampClient.ConnectedState) {
        logger.info("connected to WAMP router [{}]", client.routerUri().toASCIIString());

        subject.onNext(client);
      } else if (state instanceof WampClient.ConnectingState) {
        logger.info("connecting to WAMP router [{}]", client.routerUri().toASCIIString());

      } else if (state instanceof WampClient.DisconnectedState) {
        logger.info("disconnected from WAMP router [{}]", client.routerUri().toASCIIString());
      }
    });

    client.open();
  } catch (Exception ex) {
    logger.warn(ex.getMessage(), ex);
  }
}
 

import ws.wamp.jawampa.WampClient; //导入依赖的package包/类
public void start() {
  logger.info("Device Manager is starting...");
  try {
    WAMPClient
        .getInstance()
        .init(configuration.getAsString(Keys.WAMP_URI),
            configuration.getAsString(Keys.WAMP_REALM),
            configuration.getAsInteger(Keys.WAMP_RECONNECT),
            configuration.getAsString(Keys.WAMP_LOGIN),
            configuration.getAsString(Keys.WAMP_PASSWORD))
        .subscribe(
            (WampClient.State newState) -> {
              if (newState instanceof WampClient.ConnectedState) {
                logger.info("Connected to {}", configuration.get(Keys.WAMP_URI));
              } else if (newState instanceof WampClient.DisconnectedState) {
                logger.info("Disconnected from {}. Reason: {}",
                    configuration.get(Keys.WAMP_URI),
                    ((WampClient.DisconnectedState) newState).disconnectReason());
              } else if (newState instanceof WampClient.ConnectingState) {
                logger.info("Connecting to {}", configuration.get(Keys.WAMP_URI));
              }
            });
    logger.info("Device Proxy Service Manager started!");
  } catch (Throwable ex) {
    logger.error(ex.getMessage(), ex);
    try {
      WAMPClient.getInstance().close();
    } catch (IOException ex1) {
      logger.error(ex1.getMessage(), ex1);
    }
  }
}
 

import ws.wamp.jawampa.WampClient; //导入依赖的package包/类
@Override
public void publish(final StreamURI uri, final String body) {
    try {
        WampClient client = getOrCreateClient(uri).join();
        client.publish(uri.getTopic(), body);
    } catch (Exception ex) {
        logger.error(ex.getMessage(), ex);
    }
}
 

import ws.wamp.jawampa.WampClient; //导入依赖的package包/类
@Override
public void register(final StreamURI uri) {
    if(subscriptions.containsKey(uri)) {
        logger.debug("Stream [{}] is already being read!", uri);
        return;
    }
    try {
        WampClient client = getOrCreateClient(uri).join();
        Subscription sub = client.makeSubscription(uri.getTopic(), String.class)
                .subscribe(new ObservationConsumer(uri));
        subscriptions.put(uri, sub);
    } catch (ApplicationError ex) {
        logger.error(ex.getMessage(), ex);
    }
}
 

import ws.wamp.jawampa.WampClient; //导入依赖的package包/类
void closeIncompleteSession(Throwable disconnectReason, String optAbortReason, boolean reconnectAllowed) {
    // Send abort to the remote
    if (optAbortReason != null) {
        AbortMessage msg = new AbortMessage(null, optAbortReason);
        connectionController.sendMessage(msg, IWampConnectionPromise.Empty);
    }
    
    int nrReconnects = reconnectAllowed ? nrReconnectAttempts : 0;
    if (nrReconnects == 0) {
        stateController.setExternalState(new WampClient.DisconnectedState(disconnectReason));
    }
    WaitingForDisconnectState newState = new WaitingForDisconnectState(stateController, nrReconnects);
    connectionController.close(true, newState.closePromise());
    stateController.setState(newState);
}
 

import ws.wamp.jawampa.WampClient; //导入依赖的package包/类
@Override
public void onEnter(ClientState lastState) {
    if (lastState instanceof InitialState) {
        stateController.setExternalState(new WampClient.ConnectingState());
    }
    
    // Check for valid number of connects
    assert (nrConnectAttempts != 0);
    // Decrease remaining number of reconnects if it's not infinite
    if (nrConnectAttempts > 0) nrConnectAttempts--;
    
    // Starts an connection attempt to the router
    connectionController =
        new QueueingConnectionController(stateController.scheduler(), new ClientConnectionListener(stateController));
    
    try {
        connectingCon =
            stateController.clientConfig().connector().connect(stateController.scheduler(), this, connectionController);
    } catch (Exception e) {
        // Catch exceptions that can happen during creating the channel
        // These are normally signs that something is wrong with our configuration
        // Therefore we don't trigger retries
        stateController.setCloseError(e);
        stateController.setExternalState(new WampClient.DisconnectedState(e));
        DisconnectedState newState = new DisconnectedState(stateController, e);
        // This is a reentrant call to setState. However it works as onEnter is the last call in setState
        stateController.setState(newState);
    }
}
 

import ws.wamp.jawampa.WampClient; //导入依赖的package包/类
@Override
public void initClose() {
    reconnectSubscription.unsubscribe();
    // Current external state is Connecting
    // Move to disconnected
    stateController.setExternalState(new WampClient.DisconnectedState(null));
    // And switch the internal state also to Disconnected
    DisconnectedState newState = new DisconnectedState(stateController, null);
    stateController.setState(newState);
}
 

import ws.wamp.jawampa.WampClient; //导入依赖的package包/类
@Override
public void initClose() {
    if (nrReconnectAttempts != 0) {
        // Cancelling a reconnect triggers a state transition
        nrReconnectAttempts = 0;
        stateController.setExternalState(new WampClient.DisconnectedState(null));
    }
}
 

import ws.wamp.jawampa.WampClient; //导入依赖的package包/类
@Inject
public ServerEventDispatcher(WampClient wampClient,
                             ServerState serverState)
{
    this.wampClient = wampClient;
    this.serverState = serverState;

    this.setName(ServerEventDispatcher.class.getSimpleName());
    this.setDaemon(true);
    this.start();
}
 

import ws.wamp.jawampa.WampClient; //导入依赖的package包/类
private WampClient configureWampClient(WampRouter wampRouter)
{
    try {
        WampClient wampClient = new WampClientBuilder()
                .withRealm(WAMP_REALM)
                .withUri(INTERNAL_CLIENT_URI)
                .withConnectorProvider(new InMemoryConnectorProvider(wampRouter))
                .build();
        wampClient.open();
        return wampClient;
    }catch (Exception e) {
        throw new RuntimeException(e);
    }
}
 

import ws.wamp.jawampa.WampClient; //导入依赖的package包/类
/***
 * 
 * @param runTimeInMillis The subscription time expressed in milliseconds. The minimum runtime is 1 minute. 
 */
public void run(long runTimeInMillis)
{
    try
    {
        wampClient.statusChanged()
                .subscribe((WampClient.State newState)
                        ->
                {
                    if (newState instanceof WampClient.ConnectedState)
                    {
                        LOG.trace("Connected");

                        for (Entry<String, Action1<PubSubData>> subscription : this.subscriptions.entrySet())
                        {
                            wampClient.makeSubscription(subscription.getKey()).subscribe(subscription.getValue(), new PoloniexSubscriptionExceptionHandler(subscription.getKey()));
                        }
                    }
                    else if (newState instanceof WampClient.DisconnectedState)
                    {
                        LOG.trace("Disconnected");
                    }
                    else if (newState instanceof WampClient.ConnectingState)
                    {
                        LOG.trace("Connecting...");
                    }
                });

        wampClient.open();
        long startTime = System.currentTimeMillis();

        while (wampClient.getTerminationFuture().isDone() == false && (startTime + runTimeInMillis > System.currentTimeMillis()))
        {
            TimeUnit.MINUTES.sleep(1);
        }
    }
    catch (Exception ex)
    {
        LOG.error(("Caught exception - " + ex.getMessage()), ex);
    }
}
 

import ws.wamp.jawampa.WampClient; //导入依赖的package包/类
@Override
public void onEnter(ClientState lastState) {
    stateController.setExternalState(new WampClient.ConnectedState(sessionId, welcomeDetails, routerRoles));
}
 

import ws.wamp.jawampa.WampClient; //导入依赖的package包/类
@Override
protected void configure()
{
    // JSON configuration
    SimpleModule simpleModule = new SimpleModule();
    simpleModule.addSerializer(ExpressionValue.class, new ExpressionValueSerializer());

    ObjectMapper objectMapper = new ObjectMapper();
    objectMapper.addMixInAnnotations(VoidResult.class, VoidResultMixIn.class);
    objectMapper.addMixInAnnotations(DeferredResult.class, DeferredResultMixIn.class);
    objectMapper.addMixInAnnotations(TableResult.class, TableResultMixIn.class);
    objectMapper.addMixInAnnotations(ValueResult.class, ValueResultMixIn.class);
    objectMapper.addMixInAnnotations(TableRow.class, TableRowMixIn.class);
    objectMapper.registerModule(simpleModule);
    bind(ObjectMapper.class).toInstance(objectMapper);

    // REST API configuration
    Vertx vertx = new VertxFactoryImpl().vertx();
    bind(Vertx.class).toInstance(vertx);

    // Engine configuration
    bind(String[].class).annotatedWith(Names.named("OP_PACKAGES")).toInstance(
            new String[] {
                    "net.udidb.engine.ops.impls"
            });

    bind(DebuggeeContextManager.class).to(DebuggeeContextManagerImpl.class);

    bind(HelpMessageProvider.class).asEagerSingleton();

    bind(UdiProcessManager.class).toInstance(new UdiProcessManagerImpl());

    bind(BinaryReader.class).toInstance(new CrossPlatformBinaryReader());

    bind(ExpressionCompiler.class).toInstance(new ExpressionCompilerDelegate());

    bind(SourceLineRowFactory.class).toInstance(new InMemorySourceLineRowFactory());

    bind(ServerEngine.class).to(ServerEngineImpl.class);

    bind(OperationResultVisitor.class).to(OperationEngine.class);

    bind(ServerEventDispatcher.class).asEagerSingleton();

    bind(EventPump.class).asEagerSingleton();

    bind(EventSink.class).to(ServerEventDispatcher.class);

    WampRouter wampRouter = configureWampRouter();
    bind(WampRouter.class).toInstance(wampRouter);

    bind(WampClient.class).toInstance(configureWampClient(wampRouter));

}