import akka.actor.ReceiveTimeout; //导入依赖的package包/类
@Override
public void onReceive(Object message) {
    if (message instanceof CaptureSnapshotReply) {
        Snapshot snapshot = Snapshot.create(
                ((CaptureSnapshotReply)message).getSnapshotState(),
                params.captureSnapshot.getUnAppliedEntries(),
                params.captureSnapshot.getLastIndex(), params.captureSnapshot.getLastTerm(),
                params.captureSnapshot.getLastAppliedIndex(), params.captureSnapshot.getLastAppliedTerm(),
                params.electionTerm.getCurrentTerm(), params.electionTerm.getVotedFor(),
                params.peerInformation);

        LOG.debug("{}: Received CaptureSnapshotReply, sending {}", params.id, snapshot);

        params.replyToActor.tell(new GetSnapshotReply(params.id, snapshot), getSelf());
        getSelf().tell(PoisonPill.getInstance(), getSelf());
    } else if (message instanceof ReceiveTimeout) {
        LOG.warn("{}: Got ReceiveTimeout for inactivity - did not receive CaptureSnapshotReply within {} ms",
                params.id, params.receiveTimeout.toMillis());

        params.replyToActor.tell(new akka.actor.Status.Failure(new TimeoutException(String.format(
                "Timed out after %d ms while waiting for CaptureSnapshotReply",
                    params.receiveTimeout.toMillis()))), getSelf());
        getSelf().tell(PoisonPill.getInstance(), getSelf());
    }
}
 

import akka.actor.ReceiveTimeout; //导入依赖的package包/类
@Override
public void onReceive(Object message) {
    if (message instanceof GetSnapshotReply) {
        onGetSnapshotReply((GetSnapshotReply)message);
    } else if (message instanceof Failure) {
        LOG.debug("{}: Received {}", params.id, message);

        params.replyToActor.tell(message, getSelf());
        getSelf().tell(PoisonPill.getInstance(), getSelf());
    } else if (message instanceof ReceiveTimeout) {
        String msg = String.format(
                "Timed out after %s ms while waiting for snapshot replies from %d shard(s). %d shard(s) %s "
                + "did not respond.", params.receiveTimeout.toMillis(), params.shardNames.size(),
                remainingShardNames.size(), remainingShardNames);
        LOG.warn("{}: {}", params.id, msg);
        params.replyToActor.tell(new Failure(new TimeoutException(msg)), getSelf());
        getSelf().tell(PoisonPill.getInstance(), getSelf());
    }
}
 

import akka.actor.ReceiveTimeout; //导入依赖的package包/类
private Procedure<Object> waitForWorkIsDoneAck(final Object result) {
    return message -> {
        if (message instanceof Ack && ((Ack) message).workId.equals(jobId())) {
            sendToMaster(new MasterWorkerProtocol.WorkerRequestsWork(workerId, workerRole));
            getContext().setReceiveTimeout(Duration.Undefined());
            getContext().become(receiveBuilder()
                    .matchAny(p->idle.apply(p))
                    .build());
        } else if (message instanceof ReceiveTimeout) {
            log.info("No ack from master, retrying (" + workerId + " -> " + jobId() + ")");
            sendToMaster(new MasterWorkerProtocol.WorkIsDone(workerId, jobId(), result));
        } else {
            unhandled(message);
        }
    };
}
 

import akka.actor.ReceiveTimeout; //导入依赖的package包/类
@Override
public Receive createReceive() {
  return receiveBuilder()
    .match(ActorIdentity.class, identity -> {
      calculator = identity.getRef();
      if (calculator == null) {
        System.out.println("Remote actor not available: " + path);
      } else {
        getContext().watch(calculator);
        getContext().become(active, true);
      }
    })
    .match(ReceiveTimeout.class, x -> {
      sendIdentifyRequest();
    })
    .build();
}
 

import akka.actor.ReceiveTimeout; //导入依赖的package包/类
/**
    * After the worker sends the result of his work to the master he waits for 
    * the ACK of this message.
    * @param result
    * @return
    */
private Behavior waitForWorkIsDoneAck(final Object result) {
  return new Behavior() {
    public void apply(Object message) {
  	// The receive ACK message that mean Master receive the result of his work
      if (message instanceof Ack && ((Ack) message).workId.equals(workId())) {
      	// Worker ask for new work
        sendToMaster(new WorkerRequestsWork(workerId));
        getContext().setReceiveTimeout(Duration.Undefined());
        getContext().become(idle);
      }
      // if he does not receive ACK, worker resends a new WorkIsDone message
      else if (message instanceof ReceiveTimeout) {
        log.debug("No ack from master, retrying (" + workerId + " -> " + workId() + ")");
        sendToMaster(new WorkIsDone(workerId, workId(), result));
      }
      else {
        unhandled(message);
      }
    }
  };
}
 

import akka.actor.ReceiveTimeout; //导入依赖的package包/类
private Procedure<Object> waitingForAck() {
	return new Procedure<Object>() {

		@Override
		public void apply(Object msg) throws Exception {
			if(msg instanceof Ack) {
				log.debug("transaction committed");
				
				getContext().stop(getSelf());
			} else if(msg instanceof Failure) {
				log.error("failed to commit: {}", msg);
				
				sender.tell(msg, getSelf());
				getContext().stop(getSelf());
			} else if(msg instanceof ReceiveTimeout) {
				log.error("timout while committing");
				
				sender.tell(new Failure(new TimeoutException("while commiting")), getSelf());
				getContext().stop(getSelf());
			} else {
				unhandled(msg);
			}
		}
		
	};
}
 

import akka.actor.ReceiveTimeout; //导入依赖的package包/类
@Override
public void onReceive(Object msg) throws Exception {
	if(msg instanceof TransactionCreated) {
		log.debug("transaction created");
		
		try {
			databaseScheme = databaseConfig.getString("scheme");
		} catch(ConfigException.Missing cem) {
			databaseScheme = "SDE";
		}
		
		log.debug("database scheme before calling get fetch table: " + databaseScheme);
		
		transaction = ((TransactionCreated)msg).getActor();
		transaction.tell(SDEUtils.getFetchTable(SDEUtils.getItemsFilter(), databaseScheme), getSelf());
		
		getContext().become(onReceiveStreaming());
	} else if(msg instanceof ReceiveTimeout) {
		log.error("timeout received");
		getContext().stop(getSelf());
	} else {
		unhandled(msg);
	}
}
 

import akka.actor.ReceiveTimeout; //导入依赖的package包/类
protected Procedure<Object> onReceiveCommitAck() {
	return new Procedure<Object>() {

		@Override
		public void apply(Object msg) throws Exception {
			if(msg instanceof Ack) {
				log.debug("transaction finished");
				getContext().stop(getSelf());
			} else if(msg instanceof ReceiveTimeout) {
				log.error("timeout received");
				getContext().stop(getSelf());
			}
		}
		
	};
}
 

import akka.actor.ReceiveTimeout; //导入依赖的package包/类
private Procedure<Object> onReceiveDatasetInfo() {
	return new Procedure<Object>() {

		@Override
		public void apply(Object msg) throws Exception {
			if(msg instanceof DatasetInfo) {
				log.debug("dataset info received");
				originalSender.tell(msg, getSelf());
				transaction.tell(new Commit(), getSelf());
				getContext().become(onReceiveCommitAck());
			} else if(msg instanceof ReceiveTimeout) {
				unavailable();
			} else {
				unhandled(msg);
			}
		}
	};
}
 

import akka.actor.ReceiveTimeout; //导入依赖的package包/类
@Override
public final void onReceive(Object msg) throws Exception {
	if(msg instanceof ReceiveTimeout) {
		log.error("timeout while collecting information");
		
		sender.tell(new Failure(new TimeoutException("collecting information")), getSelf());
		getContext().stop(getSelf());
	} else if(msg instanceof MetadataNotFound) {
		log.debug("metadata not found");
		
		sender.tell(new DatasetNotFound(((MetadataNotFound)msg).getIdentification()), getSelf());
		getContext().stop(getSelf());
	} else if(msg instanceof MetadataItem) {
		log.debug("metadata item");
		
		MetadataDocumentFactory metadataDocumentFactory = new MetadataDocumentFactory();
		MetadataDocument metadataDocument = metadataDocumentFactory.parseDocument(((MetadataItem)msg).getContent());
		
		handleMetadataDocument(metadataDocument);
	} else {			
		unhandled(msg);
	}
}
 

import akka.actor.ReceiveTimeout; //导入依赖的package包/类
@Override
public void onReceive(Object msg) throws Exception {
	if(msg instanceof ReceiveTimeout) {
		log.error("timeout");
		
		sendUnavailable();
	} else if(msg instanceof MetadataNotFound) {
		log.debug("metadata not found");
		
		tellTarget(new DatasetNotFound(((MetadataNotFound) msg).getIdentification()));
	} else if(msg instanceof MetadataItem) {
		log.debug("metadata item");
			
		handleMetadataItem((MetadataItem)msg);
	}  else {
		onReceiveElse(msg);
	}
}
 

import akka.actor.ReceiveTimeout; //导入依赖的package包/类
@Override
public final void onReceive(Object msg) throws Exception {
	if(msg instanceof Commit) {
		handleCommit();
	} else if(msg instanceof Rollback) {
		handleRollback();
	} else if(msg instanceof Query) {
		handleQuery((Query)msg);
	} else if(msg instanceof StreamingQuery) {
		handleStreamingQuery((StreamingQuery)msg);
	} else if(msg instanceof ReceiveTimeout) {
		handleTimeout();
	} else if(msg instanceof Terminated) {
		handleTerminated((Terminated)msg);
	} else {
		unhandled(msg);
	}
}
 

import akka.actor.ReceiveTimeout; //导入依赖的package包/类
@Override
public void onReceive(Object msg) throws Exception {
	if(msg instanceof ReceiveTimeout) {
		log.error("timeout");
	} else {		
		log.debug("answer received");
		
		origSender.forward(msg, getContext());
		
		if(msg instanceof Failure) {
			log.debug("failure received");
		} else {
			log.debug("dispatching event messages");
			
			Event event = new Event(origMsg, msg);
			for(ActorRef listener : listeners) {
				listener.tell(event, getSelf());
			}
		}
	}
	
	getContext().stop(getSelf());
}
 

import akka.actor.ReceiveTimeout; //导入依赖的package包/类
private Procedure<Object> beforeDelete(ActorRef sender, Class<?> entity) {
	return new Procedure<Object>() {

		@Override
		public void apply(Object msg) throws Exception {
			if(msg instanceof ReceiveTimeout) {
				log.error("timeout while waiting for beforeDelete to complete");
				
				sender.tell(new EventFailed(), getSelf());
				getContext().setReceiveTimeout(Duration.Inf());
				getContext().become(receive());
			} else if(msg instanceof BeforeCompleted) {
				log.debug("before completed");
				
				sender.tell(new EventWaiting(), getSelf());
				getContext().become(afterDelete(sender, entity, ((BeforeCompleted)msg).getResult()));
			} else {
				log.debug("message stashed: {} from {}", msg, getSender());
				
				stash();
			}
		}
		
	};
}
 

import akka.actor.ReceiveTimeout; //导入依赖的package包/类
protected void become(String message, Procedure<Object> behavior) {
	log.debug("become: {}", message);
	
	getContext().become(new Procedure<Object>() {

		@Override
		public void apply(Object msg) throws Exception {
			if(msg instanceof ReceiveTimeout) {
				log.debug("receive timeout");
				
				if(!acknowledged) {
					acknowledgeJob();
				}
				
				jobContext.tell(new UpdateJobState(JobState.ABORTED), getSelf());
				
				getContext().stop(getSelf());
			} else {
				behavior.apply(msg);
			}
			
		}
	});
}
 

import akka.actor.ReceiveTimeout; //导入依赖的package包/类
@Override
public void onReceive(Object msg) throws Exception {
	if(msg instanceof ReceiveTimeout) {
		log.error("timeout while starting job: {}", jobInfo);
		
		getContext().parent().tell(new Ack(), getSelf());
		getContext().stop(getSelf());
	} else if(msg instanceof Ack) {
		log.debug("acknowledged");
		
		getContext().parent().tell(msg, getSender());
		getContext().stop(getSelf());
	} else {
		log.debug("other message");
		
		getContext().become(started());
		getSelf().forward(msg, getContext());
	}
}
 

import akka.actor.ReceiveTimeout; //导入依赖的package包/类
@Override
@SuppressWarnings("unchecked")
public void onReceive(Object msg) throws Exception {
	if(msg instanceof ReceiveTimeout) {
		log.error("timeout");
		
		getContext().stop(getSelf());
	} else if(msg instanceof DatasetInfo) {
		log.debug("dataset info received");
		
		receiver.tell(startImport((U)msg), getSelf());
	} else if(msg instanceof Ack) {
		log.debug("receiver is ready");
		
		provider.tell(getDataset(), receiver);
		
		getContext().stop(getSelf());
	} else {
		unhandled(msg);
	}
	
}
 

import akka.actor.ReceiveTimeout; //导入依赖的package包/类
private Procedure<Object> vacuuming() {
	return new Procedure<Object>() {

		@Override
		public void apply(Object msg) throws Exception {
			if(msg instanceof Vacuumed) {
				log.debug("vacuum completed");
				vacuumed(JobState.SUCCEEDED);
			} else if(msg instanceof ReceiveTimeout) {
				log.error("timeout while vacuuming");
				vacuumed(JobState.FAILED);
			} else {
				elseProvisioning(msg);
			}
		}

		private void vacuumed(JobState result) {
			getContext().parent().tell(new UpdateJobState(result), getSelf());
			
			getContext().setReceiveTimeout(Duration.Inf());
			getContext().become(receive());
		}
		
	};
}
 

import akka.actor.ReceiveTimeout; //导入依赖的package包/类
@Override
public void handleReceive(final Object message) {
    if (CloseTransaction.isSerializedType(message)) {
        closeTransaction(true);
    } else if (message instanceof ReceiveTimeout) {
        LOG.debug("Got ReceiveTimeout for inactivity - closing transaction {}", transactionId);
        closeTransaction(false);
    } else {
        unknownMessage(message);
    }
}
 

import akka.actor.ReceiveTimeout; //导入依赖的package包/类
@Override
public Receive createReceive() {
    return ReceiveBuilder.create()
        .match(CommandWithHandler.class, m -> {
            @SuppressWarnings("unchecked") CommandWithHandler msg = m;
            handleCommand(msg.command, msg.handler);
        })
        .match(commandType, this::handleCommand)
        .matchEquals(ReceiveTimeout.getInstance(), msg -> passivate())
        .match(Stop.class, msg -> context().stop(self()))
        .build();
}