import org.jetlang.fibers.Fiber; //导入依赖的package包/类
public HomeWorker(Fiber fiberThread, int mapId, int homeId, long ownerId) {
    this.fiberThread = fiberThread;
    this.mapId = mapId;
    this.homeId = homeId;
    this.ownerUid = ownerId;

    //回调方法
    Callback<List<NMessage>> callback = new HomeWorkerCallback(this);

    //像消息通道注册消息处理观察者
    BatchSubscriber<NMessage> batchSubscriber = new BatchSubscriber<NMessage>(this.fiberThread, callback, 10, TimeUnit.MILLISECONDS);
    this.messageChannel.subscribe(this.fiberThread, batchSubscriber);

    //改变房间状态为玩家准备中
    this.status = HomeStatusEnum.PLAYER_READY_DOING;
}
 

import org.jetlang.fibers.Fiber; //导入依赖的package包/类
public Service(String name) {
	serviceName = name;
	Fiber fiber = Kernel.getInstance().getScheduler().newFiber();
	fiber.start();
	Callback<Packet> sysCtrlCallback = new Callback<Packet>() {
		public void onMessage(Packet pkt) {
			if(pkt instanceof ControlPacket){
				processSysCtrl((ControlPacket) pkt);
			}
		}
	};
	subscription = new Subscription(CoreChannel.SYSCTRL, fiber, sysCtrlCallback);
	
	try {
		Kernel.getInstance().getStream().subscribe(subscription);
	} catch (CommunicationException e) {
		System.err.println(e.getMessage());
	}
	
	taskRegistry = new FastMap<String, Taskable>();
}
 

import org.jetlang.fibers.Fiber; //导入依赖的package包/类
public Log4jService() {
	super(Log4jService.class.getSimpleName());
	PropertyConfigurator.configure(CFG_FILE);
	Fiber fiber = Kernel.getInstance().getScheduler().newFiber();
	fiber.start();
	Callback<Packet> callback = new Callback<Packet>() {
		public void onMessage(Packet packet) {
			LogPacket p = (LogPacket) packet;
			Logger.getLogger(p.getSource()).log(Level.toLevel(p.getLevel()), p.getMessage());
		}
	};
	
	subscription = new Subscription(CoreChannel.LOG, fiber, callback);
	
	try {
		Kernel.getInstance().getStream().subscribe(subscription);
	} catch (CommunicationException e) {
		e.printStackTrace();
	}
}
 

import org.jetlang.fibers.Fiber; //导入依赖的package包/类
public TwitterService(Properties properties) {
	super(TwitterService.class.getSimpleName());
	tweeter = new Tweeter(properties);
	Fiber fiber = Kernel.getInstance().getScheduler().newFiber();
	fiber.start();
	
	Callback<Packet> callback = new Callback<Packet>() {
		public void onMessage(Packet pkt) {
			LogPacket l = (LogPacket) pkt;
			if(l.getSource().equals("twitter")) {
				tweeter.tweet(l.getMessage());
			}
		}
	};
	
	subscription = new Subscription(CoreChannel.LOG, fiber, callback);
	
	try {
		Kernel.getInstance().getStream().subscribe(subscription);
	} catch (CommunicationException e) {
		// TODO Auto-generated catch block
		e.printStackTrace();
	}
}
 

import org.jetlang.fibers.Fiber; //导入依赖的package包/类
private void whenTheReplicatorIsInState(Replicator.State state) throws Exception {
  final Fiber replicatorFiber = new ThreadFiber(new RunnableExecutorImpl(batchExecutor),
      "replicatorFiber-Thread", true);

  replicatorInstance = new ReplicatorInstance(replicatorFiber,
      MY_ID,
      QUORUM_ID,
      log,
      clock,
      persistence,
      sendRpcChannel,
      eventChannel,
      commitNotices,
      state);

  sendRpcChannel.subscribe(rpcFiber, (request) -> {
    if (request.getRequest().to == MY_ID) {
      handleLoopBack(request);
    }
  });

  replicatorInstance.start();
}
 

import org.jetlang.fibers.Fiber; //导入依赖的package包/类
public static <V> void addCallback(@NotNull ListenableFuture<V> future,
                                   @NotNull Consumer<? super V> success,
                                   @NotNull Consumer<Throwable> failure,
                                   @NotNull Fiber fiber) {
  Runnable callbackListener = () -> {
    final V value;
    try {
      value = getUninterruptibly(future);
    } catch (ExecutionException | RuntimeException | Error e) {
      failure.accept(e);
      return;
    }
    success.accept(value);
  };
  future.addListener(callbackListener, fiber);
}
 

import org.jetlang.fibers.Fiber; //导入依赖的package包/类
public static <S, R> ChannelListener<S> waitForReply(RequestChannel<S, R> channel) {
  List<Throwable> throwables = new ArrayList<>();
  BatchExecutor exceptionHandlingBatchExecutor = new ExceptionHandlingBatchExecutor(throwables::add);
  RunnableExecutor runnableExecutor = new RunnableExecutorImpl(exceptionHandlingBatchExecutor);
  Fiber channelSubscriberFiber = new ThreadFiber(runnableExecutor, null, true);
  ArrayBlockingQueue<S> messages = new ArrayBlockingQueue<>(1);
  channel.subscribe(channelSubscriberFiber, m -> {
    try {
      messages.put(m.getRequest());
    } catch (InterruptedException e) {
      throw new RuntimeException(e);
    }
  });
  channelSubscriberFiber.start();
  return new ChannelListener<>(channelSubscriberFiber, messages, throwables);
}
 

import org.jetlang.fibers.Fiber; //导入依赖的package包/类
public static <T> ChannelListener<T> listenTo(Subscriber<T> channel) {
  List<Throwable> throwables = new ArrayList<>();
  BatchExecutor exceptionHandlingBatchExecutor = new ExceptionHandlingBatchExecutor(throwables::add);
  RunnableExecutor runnableExecutor = new RunnableExecutorImpl(exceptionHandlingBatchExecutor);
  Fiber channelSubscriberFiber = new ThreadFiber(runnableExecutor, null, true);
  ArrayBlockingQueue<T> messages = new ArrayBlockingQueue<>(1);
  channel.subscribe(channelSubscriberFiber, m -> {
    try {
      messages.put(m);
    } catch (InterruptedException e) {
      throw new RuntimeException(e);
    }
  });
  channelSubscriberFiber.start();
  return new ChannelListener<>(channelSubscriberFiber, messages, throwables);
}
 

import org.jetlang.fibers.Fiber; //导入依赖的package包/类
public SingleReplicatorController(int port,
                                  long nodeId,
                                  Collection<Long> peerIds,
                                  Path baseTestPath,
                                  Fiber testFiber,
                                  FiberSupplier fiberSupplier,
                                  Consumer<Throwable> exceptionHandler,
                                  EventLoopGroup bossGroup,
                                  EventLoopGroup workerGroup) throws Exception {

  moduleInfo = new SimpleModuleInformationProvider(testFiber, exceptionHandler);

  replicationModule =
      new ReplicatorService(bossGroup, workerGroup, nodeId, port, moduleInfo, fiberSupplier,
          new NioQuorumFileReaderWriter(baseTestPath));
  logModule = new LogService(baseTestPath, fiberSupplier);
  nodeInfoModule = new BeaconService(nodeId, DISCOVERY_PORT, workerGroup, moduleInfo, fiberSupplier);

  startAll();

  service = new C5GeneralizedReplicationService(replicationModule, logModule, fiberSupplier);
  replicator = service.createReplicator(QUORUM_ID, peerIds).get();
}
 

import org.jetlang.fibers.Fiber; //导入依赖的package包/类
@Test
public void writeOutAsynchronously() throws ExecutionException, InterruptedException, TimeoutException {
  RegionSpecifier regionSpecifier = new RegionSpecifier(RegionSpecifier.RegionSpecifierType.REGION_NAME,
      ByteBuffer.wrap(Bytes.toBytes("c5:writeOutAsynchronously")));
  ByteBuffer cq = ByteBuffer.wrap(Bytes.toBytes("cq"));
  ByteBuffer cf = ByteBuffer.wrap(Bytes.toBytes("cf"));
  ByteBuffer value = ByteBuffer.wrap(new byte[512]);
  MutationProto.ColumnValue.QualifierValue qualifierValue = new MutationProto.ColumnValue.QualifierValue(cq,
      value,
      0l,
      null);

  SingleNodeTableInterface singleNodeTable = new SingleNodeTableInterface("localhost", getRegionServerPort());

  Fiber flusher = new ThreadFiber();
  flusher.start();
  flusher.scheduleAtFixedRate(singleNodeTable::flushHandler, 50, 50, MILLISECONDS);

  for (int i = 0; i != TO_SEND; i++) {
    sendProto(regionSpecifier, cf, Arrays.asList(qualifierValue), singleNodeTable, i);
  }
  System.out.println("buffered");
  countDownLatch.await();
}
 

import org.jetlang.fibers.Fiber; //导入依赖的package包/类
private void scan(ChannelHandlerContext ctx, Call call) throws IOException,
    RegionNotFoundException {
  final ScanRequest scanIn = call.getScan();
  if (scanIn == null) {
    throw new IOException("Poorly specified scan. There is no actual get data in the RPC");
  }

  final long scannerId;
  scannerId = getScannerId(scanIn);
  final Integer numberOfRowsToSend = scanIn.getNumberOfRows();
  Channel<Integer> channel = scanManager.getChannel(scannerId);
  // New Scanner
  if (null == channel) {
    final Fiber fiber = new ThreadFiber();
    fiber.start();
    channel = new MemoryChannel<>();
    Region region = regionServerService.getOnlineRegion(call.getScan().getRegion());
    final ScanRunnable scanRunnable = new ScanRunnable(ctx, call, scannerId, region);
    channel.subscribe(fiber, scanRunnable);
    scanManager.addChannel(scannerId, channel);
  }
  channel.publish(numberOfRowsToSend);
}
 

import org.jetlang.fibers.Fiber; //导入依赖的package包/类
/** Creates a JetlangPonger. */
public JetlangPonger(final Fiber _fiber) {
    fiber = _fiber;
    channel = new MemoryRequestChannel();
    final Callback<Request> onReq = new Callback<Request>() {
        @Override
        public void onMessage(final Request message) {
            final Object request = message.getRequest();
            if (request instanceof PingRequest) {
                final PingRequest ping = (PingRequest) request;
                message.reply(ping.processRequest(JetlangPonger.this));
            } else {
                throw new IllegalStateException(
                        "Expected PingRequest but got "
                                + request.getClass());
            }
        }
    };
    fiber.start();
    channel.subscribe(fiber, onReq);
}
 

import org.jetlang.fibers.Fiber; //导入依赖的package包/类
public JATOInstanceActor(JATOActorFactory actorFactory, T instance, Channel<MethodMessage> mailbox, Fiber fiber) {
    super(actorFactory, mailbox, fiber);
    this.instance = instance;
    this.clazz = (Class<T>) instance.getClass();
    methodAccess = MethodAccess.get(clazz);
    methodNameMap = buildMethodNameMap(methodAccess);
}
 

import org.jetlang.fibers.Fiber; //导入依赖的package包/类
public JATOAbstractActor(JATOActorFactory actorFactory, Channel<T> mailbox, Fiber fiber) {
    this.actorFactory = actorFactory;
    this.mailbox = mailbox;
    this.fiber = fiber;
    final JATOAbstractActor<T> self = this;
    this.callback = new Callback<T>() {
        public void onMessage(T t) {
            getActorFactory().addInstanceActorToCache(self, self);
            self.onReceive(t);
        }
    };
    mailbox.subscribe(fiber, callback);
}
 

import org.jetlang.fibers.Fiber; //导入依赖的package包/类
public <T> JATOCallbackActor<T> startCallbackActor(Callback<T> callback) {
    Fiber fiber = startFiber();
    Channel<T> mailbox = new MemoryChannel<T>();
    JATOCallbackActor<T> actor = new JATOCallbackActor<T>(this, mailbox, fiber, callback);
    return actor;

}
 

import org.jetlang.fibers.Fiber; //导入依赖的package包/类
private Replicator createReplicatorWithLog(ReplicatorLog log, String quorumId, Collection<Long> peers) {
  if (replicatorInstances.containsKey(quorumId)) {
    LOG.debug("Replicator for quorum {} exists already", quorumId);
    return replicatorInstances.get(quorumId);
  }

  if (!peers.contains(nodeId)) {
    LOG.warn("Creating a replicator instance for quorum {} peers {} but it does not contain me ({})",
        quorumId, peers, nodeId);
  }

  LOG.info("Creating replicator instance for {} peers {}", quorumId, peers);

  MemoryChannel<Throwable> throwableChannel = new MemoryChannel<>();
  Fiber instanceFiber = fiberSupplier.getNewFiber(throwableChannel::publish);
  ReplicatorInstance instance =
      new ReplicatorInstance(
          instanceFiber,
          nodeId,
          quorumId,
          log,
          new DefaultSystemTimeReplicatorClock(),
          persister,
          outgoingRequests,
          replicatorEventChannel,
          indexCommitNotices,
          Replicator.State.FOLLOWER
      );
  if (log.getLastIndex() == 0) {
    instance.bootstrapQuorum(peers);
  }
  throwableChannel.subscribe(fiber, instance::failReplicatorInstance);
  replicatorInstances.put(quorumId, instance);
  instance.start();
  return instance;
}
 

import org.jetlang.fibers.Fiber; //导入依赖的package包/类
/**
 * Both the fiber and replicator must be started by the user of this class, and the
 * user takes responsibility for their disposal.
 */
public C5GeneralizedReplicator(Replicator replicator, Fiber fiber) {
  this.nodeId = replicator.getId();
  this.replicator = replicator;
  this.fiber = fiber;

  setupCommitNoticeSubscription();
  setupEventNoticeSubscription();
}
 

import org.jetlang.fibers.Fiber; //导入依赖的package包/类
@Before
public final void createLeaderAndSetupFibersAndChannels() throws Exception {
  sendRpcChannel.subscribe(rpcFiber, (request) -> System.out.println(request.getRequest()));
  sendRpcChannel.subscribe(rpcFiber, this::routeOutboundRequests);
  sendRpcChannel.subscribe(rpcFiber, requestLog::publish);

  Fiber replicatorFiber = new ThreadFiber(new RunnableExecutorImpl(batchExecutor), "replicatorFiber-Thread", true);
  InRamSim.StoppableClock clock = new InRamSim.StoppableClock(0, 1000);
  clock.startTimeout();

  log.logEntries(
      Lists.newArrayList(
          new LogEntry(CURRENT_TERM, 1, new ArrayList<>(), QuorumConfiguration.of(PEER_ID_LIST).toProtostuff())));
  lastIndex = 1;

  ReplicatorInfoPersistence persister = new InRamSim.Persister();
  persister.writeCurrentTermAndVotedFor(QUORUM_ID, CURRENT_TERM, LEADER_ID);

  replicatorInstance = new ReplicatorInstance(replicatorFiber,
      LEADER_ID,
      QUORUM_ID,
      log,
      clock,
      new InRamSim.Persister(),
      sendRpcChannel,
      new MemoryChannel<>(),
      commitNotices,
      State.LEADER);
  replicatorInstance.start();
  rpcFiber.start();
}
 

import org.jetlang.fibers.Fiber; //导入依赖的package包/类
public QuorumOfReplicatorsController(Collection<Long> peerIds,
                                     Path baseTestPath,
                                     Fiber testFiber,
                                     FiberSupplier fiberSupplier,
                                     Consumer<Throwable> exceptionHandler) throws Exception {

  this.peerIds = peerIds;
  this.baseTestPath = baseTestPath;
  this.testFiber = testFiber;
  this.fiberSupplier = fiberSupplier;
  this.exceptionHandler = exceptionHandler;

  createControllersForEachPeerId();
}
 

import org.jetlang.fibers.Fiber; //导入依赖的package包/类
@BeforeClass
public static void beforeClass() throws Exception {
  if (dirty) {
    System.setProperty(C5ServerConstants.C5_CFG_PATH, ClusterOrPseudoCluster.testFolder.getRoot().getAbsolutePath());
    System.setProperty("clusterName", C5ServerConstants.LOCALHOST);

    server = Main.startC5Server(new String[]{});
    metaOnNode = server.getNodeId();
    TabletModule tabletServer = (TabletModule) server.getModule(ModuleType.Tablet).get();
    RegionServerModule regionServer = (RegionServerModule) server.getModule(ModuleType.RegionServer).get();
    stateChanges = tabletServer.getTabletStateChanges();

    Fiber receiver = new ThreadFiber(new RunnableExecutorImpl(), "cluster-receiver-static-fiber", false);
    receiver.start();

    // create java.util.concurrent.CountDownLatch to notify when message arrives
    final CountDownLatch latch = new CountDownLatch(1);
    Fiber fiber = new ThreadFiber(new RunnableExecutorImpl(), "cluster-tablet-state-change-fiber", false);
    fiber.start();
    tabletServer.getTabletStateChanges().subscribe(fiber, tabletStateChange -> {
      if (tabletStateChange.state.equals(Tablet.State.Leader)) {
        if (tabletStateChange.tablet.getRegionInfo().getRegionNameAsString().startsWith("hbase:meta")) {
          metaOnPort = regionServer.port();
          metaOnNode = server.getNodeId();

          latch.countDown();
          fiber.dispose();
        }
      }
    });

    latch.await();
    receiver.dispose();
  }
  dirty = false;
}