WAMP-网络程序消息协议

package ws.wamp.jawampa;

import java.io.IOException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import rx.Scheduler;
import rx.Subscription;
import rx.functions.Action0;
import rx.functions.Action1;
import rx.schedulers.Schedulers;
import ws.wamp.jawampa.ApplicationError;
import ws.wamp.jawampa.Request;
import ws.wamp.jawampa.WampError;
import ws.wamp.jawampa.WampClient;
import ws.wamp.jawampa.WampClientBuilder;

public class Publish {
    
    String url;
    String realm;
    int flag = 0;
    WampClient client;
    
    Subscription addProcSubscription;
    Subscription counterPublication;
    Subscription onHelloSubscription;
    
    // Scheduler for this example
    ExecutorService executor = Executors.newSingleThreadExecutor();
    Scheduler rxScheduler = Schedulers.from(executor);
    
    static final int TIMER_INTERVAL = 1000; // 1s
    int counter = 0;
    
    Publish(String url, String realm) throws Exception {
        this.url = url;
        this.realm = realm;
    }
    
    
    public static void main(String[] args) throws Exception {
      if (args.length != 2) {
          System.out.println("Need 2 commandline arguments: Router URL and ream name");
          return;
      }
      String args1 ="ws://172.16.100.55:8080/ws";
      String args2 = "realm1";
      new Publish(args1, args2).run();
  }
  
  void run() {
      
      WampClientBuilder builder = new WampClientBuilder();
      try {
          builder.witUri(url)
                 .withRealm(realm)
                 .withInfiniteReconnects()
                 .withCloseOnErrors(true)
                 .withReconnectInterval(5, TimeUnit.SECONDS);
          client = builder.build();
      } catch (WampError e) {
          e.printStackTrace();
          return;
      }

      // Subscribe on the clients status updates
      client.statusChanged()
            .observeOn(rxScheduler)
            .subscribe(new Action1<WampClient.Status>() {
          @Override
          public void call(WampClient.Status t1) {
              System.out.println("Session status changed to " + t1);

              if (t1 == WampClient.Status.Connected) {
                  
               
                  // PUBLISH and CALL every second .. forever
                  counter = 0;
                  final int published = 4;
                  onHelloSubscription= client.publish("glf", published)
                                .observeOn(rxScheduler)
                                .subscribe(new Action1<Long>() {
                                  @Override
                                  public void call(Long t1) {
                                      System.out.println("published to ‘oncounter‘ with counter " + published);
                                  }
                              }, new Action1<Throwable>() {
                                  @Override
                                  public void call(Throwable e) {
                                      System.out.println("Error during publishing to ‘oncounter‘: " + e);
                                  }
                              });
                          
                           CALL a remote procedure
                            
                              flag =1;
                               System.out.println(flag);
                          counter++;
                     
              }
              else if (t1 == WampClient.Status.Disconnected) {
                  closeSubscriptions();
              }
          }
      }, new Action1<Throwable>() {
          @Override
          public void call(Throwable t) {
              System.out.println("Session ended with error " + t);
          }
      }, new Action0() {
          @Override
          public void call() {
              System.out.println("Session ended normally");
          }
      });

      client.open();
      
      waitUntilKeypressed();
      while (flag == 0){
          try {
              Thread.sleep(1000);
          } catch (InterruptedException e) {
              
              e.printStackTrace();
          }
          
          System.out.println("waiting");
          }
      System.out.println("Shutting down");
      closeSubscriptions();
      client.close();
      try {
          client.getTerminationFuture().get();
      } catch (Exception e) {}
      
      executor.shutdown();
  }
  
  void closeSubscriptions() {
      if (onHelloSubscription != null)
          onHelloSubscription.unsubscribe();
      onHelloSubscription = null;
      if (counterPublication != null)
          counterPublication.unsubscribe();
      counterPublication = null;
      if (addProcSubscription != null)
          addProcSubscription.unsubscribe();
      addProcSubscription = null;
  }
  
  private void waitUntilKeypressed() {
      try {
          System.in.read();
          while (System.in.available() > 0) {
              System.in.read();
          }
      } catch (IOException e) {
          e.printStackTrace();
      }
  }

}