import org.jetlang.core.Callback; //导入依赖的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.core.Callback; //导入依赖的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.core.Callback; //导入依赖的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.core.Callback; //导入依赖的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.core.Callback; //导入依赖的package包/类
public ControlTester(){
	counter = 0;
	setDelay(1000l);
	getRequiredDevices().add("localpose");

	Callback<Packet> myCbk = new Callback<Packet>(){

		@Override
		public void onMessage(Packet message) {
			if(message instanceof LocalPosePacket){
				Kernel.getInstance().getSyslog().debug(this.getClass().getSimpleName() + " received localpose data.");
			}

		}

	};
	subscribe(CoreChannel.SYSTEM, myCbk);
}
 

import org.jetlang.core.Callback; //导入依赖的package包/类
@SuppressWarnings("unchecked")
	public Device(Backend backend, String type, String channel, long delay) {
		super();
		setDelay(delay);
		if(channel != null){
			Callback<Packet> cbk = new Callback<Packet>(){

				public void onMessage(Packet pkt) {
					if(pkt.getPacketType().equals(device))
						driver.request((T) pkt);
				}
				
			};
			subscribe(channel, cbk);
		}
//		super(channel, delay);
		driver = (HardwareDriver<T>) backend.getDriver(type + "Driver");
		if(driver == null) {
			Kernel.getInstance().getSyslog().error("Failed to load " + type + " driver.");
		}
		device = type.toLowerCase();
	}
 

import org.jetlang.core.Callback; //导入依赖的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.core.Callback; //导入依赖的package包/类
/** Send a ping, unless we're done, in which case tell caller we are done. */
private void ping(final Request hammerRequest) throws Exception {
    final Callback<Integer> onReply = new Callback<Integer>() {
        @Override
        public void onMessage(final Integer response) {
            pongs++;
            if (response.intValue() != pongs) {
                throw new IllegalStateException("Expected " + pongs
                        + " but got " + response);
            }
            if (pongs < count) {
                try {
                    ping(hammerRequest);
                } catch (final Exception e) {
                    e.printStackTrace();
                    hammerRequest.reply(pongs);
                }
            } else {
                hammerRequest.reply(pongs);
            }
        }
    };
    ponger.ping(pongs, onReply);
}
 

import org.jetlang.core.Callback; //导入依赖的package包/类
/** Tells the pinger to hammer the Ponger. Blocks and returns the result. */
@SuppressWarnings({ "rawtypes", "unchecked" })
public Integer hammer(final JetlangPonger ponger, final int _count)
        throws Exception {
    final AtomicReference response = new AtomicReference();
    final HammerRequest req = new HammerRequest(ponger, _count);
    final CountDownLatch done = new CountDownLatch(1);
    final Callback onReply = new Callback() {
        @Override
        public void onMessage(final Object message) {
            // TODO Can we safely access the pinger's state from here?!?
            response.set(message);
            done.countDown();
        }
    };
    AsyncRequest.withOneReply(fiber, channel, req, onReply);
    done.await(60, TimeUnit.SECONDS);
    return (Integer) response.get();
}
 

import org.jetlang.core.Callback; //导入依赖的package包/类
public AsyncProcessor() {
    ExecutorService exec = Executors.newCachedThreadPool();
    PoolFiberFactory factory = new PoolFiberFactory(exec);

    this.fiberThread = factory.create();

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

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

import org.jetlang.core.Callback; //导入依赖的package包/类
/**
 * 初始化
 */
public void init() {
    //回调方法
    Callback<List<InvokeFun>> callback = new ClientManagerCallback(this);

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

    this.fiberThread.start();
}
 

import org.jetlang.core.Callback; //导入依赖的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.core.Callback; //导入依赖的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.core.Callback; //导入依赖的package包/类
public GoToGoal(){
		setDelay(1000l);

		goalX = 0;//Kernel.homeX;
		goalY = 0;//Kernel.homeY;

		Kernel.getInstance().getSyslog().debug("Going home: " + goalX + ", " + goalY);

		Callback<Packet> localPoseCbk = new Callback<Packet>(){

			@Override
			public void onMessage(Packet pkt) {

//				Kernel.getInstance().getSyslog().debug("Got message: " + pkt.getPacketType());

				if(pkt.getPacketType().equals(PacketType.LOCAL_POSE)){
					try{
						float xPos = ((LocalPosePacket) pkt).getPositionX();
						float yPos = ((LocalPosePacket) pkt).getPositionY();
						float theta = ((LocalPosePacket) pkt).getTheta();

						MotorPacket ctrl = calculateControl(xPos, yPos, theta);
						publish(CoreChannel.CTRL, ctrl);
					} catch(Exception e){
						e.printStackTrace();
					}
				}

			}

		};
		subscribe(CoreChannel.SYSTEM, localPoseCbk);
		// zero out motors
		publish(CoreChannel.CTRL, new MotorPacket());
	}
 

import org.jetlang.core.Callback; //导入依赖的package包/类
public JoystickControl() {
	setDelay(200);
	Callback<Packet> jsCbk = new Callback<Packet>() {
		
		public void onMessage(Packet message) {
			if(message instanceof JoystickPacket) {
				jsPkt = (JoystickPacket) message;
				//System.out.println("Got joystick!");
			}
		}
	};
	subscribe(CoreChannel.SYSTEM, jsCbk);
}
 

import org.jetlang.core.Callback; //导入依赖的package包/类
public void subscribe(String chl, Callback<Packet> cbk) {
	// Changed this method to use Callback<Packet> to make it type safe!
	if(chl != null){
		Subscription s = new Subscription(chl, taskFiber, cbk);
		subscriptions.add(s);
		try {
			Kernel.getInstance().getStream().subscribe(s);
		} catch (CommunicationException e) {
			Kernel.getInstance().getSyslog().error(e.getMessage());
		}
		if(subscriptions.size() > 0){
			this.isEventDriven = true;
		}
	}
}
 

import org.jetlang.core.Callback; //导入依赖的package包/类
/**
 * Execute a callback on all pending requests to a given peer, and any requests hereafter
 */
private void createRequestRule(long peerId, Callback<Request<RpcRequest, RpcWireReply>> handler) {
  peerBehavioralCallbacks.put(peerId, handler);
  if (requests.containsKey(peerId)) {
    List<Request<RpcRequest, RpcWireReply>> pendingRequests = new LinkedList<>();
    requests.get(peerId).drainTo(pendingRequests);
    pendingRequests.forEach(handler::onMessage);
  }
}
 

import org.jetlang.core.Callback; //导入依赖的package包/类
@Before
public void before() throws InterruptedException, ExecutionException, TimeoutException {
  Fiber receiver = new ThreadFiber(new RunnableExecutorImpl(), "cluster-receiver-fiber", false);
  receiver.start();

  final CountDownLatch latch = new CountDownLatch(1);
  Callback<TabletStateChange> onMsg = message -> {
    if (message.state.equals(Tablet.State.Leader)) {
      latch.countDown();
    }
  };
  stateChanges.subscribe(receiver, onMsg);

  TableName clientTableName = TabletNameHelpers.getClientTableName("c5", name.getMethodName());
  org.apache.hadoop.hbase.TableName tableName = TabletNameHelpers.getHBaseTableName(clientTableName);
  Channel<CommandRpcRequest<?>> commandChannel = server.getCommandChannel();

  ModuleSubCommand createTableSubCommand = new ModuleSubCommand(ModuleType.Tablet,
      TestHelpers.getCreateTabletSubCommand(tableName, splitkeys, Arrays.asList(server)));
  CommandRpcRequest<ModuleSubCommand> createTableCommand = new CommandRpcRequest<>(server.getNodeId(),
      createTableSubCommand);

  commandChannel.publish(createTableCommand);
  latch.await();

  table = new FakeHTable(C5TestServerConstants.LOCALHOST, getRegionServerPort(),
      TabletNameHelpers.toByteString(clientTableName));
  row = Bytes.toBytes(name.getMethodName());
  receiver.dispose();
}
 

import org.jetlang.core.Callback; //导入依赖的package包/类
public InRamSim(final int peerSize) {
    this.peerSize = peerSize;
    this.fiberPool = new PoolFiberFactory(Executors.newCachedThreadPool());
    Random r = new Random();

    for (int i = 0; i < peerSize; i++) {
        peerUUIDs.add((long)r.nextInt());
    }

    long plusMillis = 0;
    for( long peerId : peerUUIDs) {
        // make me a ....
        FleaseLease fl = new FleaseLease(fiberPool.create(), new Info(plusMillis), ""+peerId, "lease", peerId, peerUUIDs, rpcChannel);
        fleaseRunners.put(peerId, fl);
        plusMillis += 500;
    }

    rpcFiber = fiberPool.create();


    // subscribe to the rpcChannel:
    rpcChannel.subscribe(rpcFiber, new Callback<Request<OutgoingRpcRequest, IncomingRpcReply>>() {
        @Override
        public void onMessage(Request<OutgoingRpcRequest, IncomingRpcReply> message) {
            messageForwarder(message);
        }
    });

    rpcFiber.start();
}
 

import org.jetlang.core.Callback; //导入依赖的package包/类
private void messageForwarder(final Request<OutgoingRpcRequest, IncomingRpcReply> origMsg) {

        // ok, who sent this?!!!!!
        final OutgoingRpcRequest request = origMsg.getRequest();
//        msgSize.update(request.message.getSerializedSize());
        final long dest = request.to;
        // find it:
        final FleaseLease fl = fleaseRunners.get(dest);
        if (fl == null) {
            // boo
            LOG.error("Request to non exist: " + dest);
            origMsg.reply(null);
            return;
        }

        messages.mark();
        messageTxn.inc();

        //LOG.debug("Forwarding message from {} to {}, contents: {}", request.from, request.to, request.message);
        // Construct and send a IncomingRpcRequest from the OutgoingRpcRequest.
        // There is absolutely no way to know who this is from at this point from the infrastructure.
        final IncomingRpcRequest newRequest = new IncomingRpcRequest(1, request.from, request.message);
        AsyncRequest.withOneReply(rpcFiber, fl.getIncomingChannel(), newRequest, new Callback<OutgoingRpcReply>() {
            @Override
            public void onMessage(OutgoingRpcReply msg) {
                // Translate the OutgoingRpcReply -> IncomingRpcReply.
                //LOG.debug("Forwarding reply message from {} back to {}, contents: {}", dest, request.to, msg.message);
                messages.mark();
                messageTxn.dec();
//                msgSize.update(msg.message.getSerializedSize());
                IncomingRpcReply newReply = new IncomingRpcReply(msg.message, dest);
                origMsg.reply(newReply);
            }
        });
    }