Java 类akka.actor.ReceiveTimeout 实例源码

@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());
    }
}

@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());
    }
}

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);
        }
    };
}

@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();
}

/**
    * 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);
      }
    }
  };
}

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);
            }
        }

    };
}

@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);
    }
}

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());
            }
        }

    };
}

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);
            }
        }
    };
}

@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);
    }
}

@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);
    }
}

@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);
    }
}

@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());
}

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();
            }
        }

    };
}

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);
            }

        }
    });
}

@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());
    }
}

@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);
    }

}

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());
        }

    };
}

@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);
    }
}

@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();
}

@Override
public Receive createReceive() {
    return receiveBuilder()
        .match(String.class, key -> handleRequest())
        .match(ReceiveTimeout.class, msg -> passivate())
        .match(Stop.class, msg -> context().stop(self()))
        .build();
}

@Override
public void onReceive(final Object message) throws Exception {
    logger.info("ReceiveTimeoutUntypedActor:{}", message);
    if (message.equals("Timeout")) {

        getContext().setReceiveTimeout(Duration.create(1, TimeUnit.SECONDS));
    } else if (message instanceof ReceiveTimeout) {
        getContext().setReceiveTimeout(Duration.Undefined());
    } else {
        unhandled(message);
    }
}

private Procedure<Object> fetchingData(ActorRef transaction) {
    return new Procedure<Object>() {

        @Override
        public void apply(Object msg) throws Exception {
            if(msg instanceof ReceiveTimeout) {
                log.error("timeout while fetching data");

                sender.tell(new Failure(new TimeoutException("while fetching data")), getSelf());
                getContext().stop(getSelf());
            } else if(msg instanceof TableNotFound) {
                log.debug("table not found");

                sender.tell(new DatasetNotAvailable(request.getIdentification()), getSelf());
                getContext().stop(getSelf());
            } else if(msg instanceof Item) {
                log.debug("item");

                sender.tell(msg, getSender());                  

                // we assume that the the database is still
                // producing records as long as the cursor actor
                // is still alive.
                getContext().watch(getSender());

                // disable the receive timeout as we don't get to see
                // additional records objects as these are send directly
                // to the consumer. 
                getContext().setReceiveTimeout(Duration.Inf());  
            } else if(msg instanceof Terminated) {
                log.debug("cursor terminated");

                transaction.tell(new Commit(), getSelf());
                getContext().become(waitingForAck());
            } else {
                unhandled(msg);
            }
        }           
    };
}

private Procedure<Object> onReceiveItemRecords() {
    return new Procedure<Object>() {

        @Override
        public void apply(Object msg) throws Exception {
            if(msg instanceof SDEItemInfo) {
                log.debug("item info received");

                SDEItemInfo itemInfo = (SDEItemInfo)msg;
                SDEItemInfoType type = itemInfo.getType();
                if(SDEItemInfoType.RASTER_DATASET == type) {
                    Path file = Paths.get(itemInfo.getPhysicalname() + ".tif");

                    log.debug("fetching file: {}", file);
                    rasterFolder.tell(new FetchFile(file), originalSender);
                    getContext().stop(getSelf());
                } else {
                    log.error("wrong item type: {}", type);
                    unavailable();
                }
            } else if(msg instanceof ReceiveTimeout) {
                log.debug("timeout received");
                unavailable();
            }
        }
    };
}

@Override
public void onReceive(Object msg) throws Exception {
    if(msg instanceof TransactionCreated) {
        transaction = ((TransactionCreated)msg).getActor();
        ActorRef recordsReceiver = getContext().actorOf(
            SDEReceiveSingleItemInfo.props(getSelf()), 
            "item-records-receiver");

        try {
            databaseScheme = databaseConfig.getString("scheme");
        } catch(ConfigException.Missing cem) {
            databaseScheme = "SDE";
        }

        log.debug("database scheme before calling get fetch table: " + databaseScheme);

        transaction.tell(
            SDEUtils.getFetchTable(SDEUtils.getItemsFilter(originalMsg.getIdentification()), databaseScheme),
            recordsReceiver);
        getContext().become(onReceiveItemRecords());
    } else if(msg instanceof ReceiveTimeout) {
        log.error("timeout received");
        unavailable();
    } else {
        unhandled(msg);
    }
}

private 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());
            }
        }
    };
}

private Procedure<Object> onReceiveItems() {
    return new Procedure<Object>() {

        ActorRef consumer = originalSender;

        ActorRef producer;

        @Override
        public void apply(Object msg) throws Exception {
            if(msg instanceof Item) {
                log.debug("item");
                producer = getSender();
                consumer.tell(msg, getSelf());
            } else if(msg instanceof NextItem) {
                log.debug("next");
                consumer = getSender();
                producer.tell(msg, getSelf());
            } else if(msg instanceof Stop) {
                log.warning("stop");
                producer.tell(msg, getSender());
            } else if(msg instanceof End) {
                log.debug("end");
                consumer.tell(msg, getSelf());
                transaction.tell(new Commit(), getSelf());
                getContext().become(onReceiveCommitAck());
            } else if(msg instanceof ReceiveTimeout) {
                log.error("timeout received");
                consumer.tell(new Unavailable(), getSelf());
                transaction.tell(new Commit(), getSelf());
                getContext().become(onReceiveCommitAck());
            } else {
                unhandled(msg);
            }
        }

    };
}

private Procedure<Object> onReceiveDatabaseTableInfo() {
    return new Procedure<Object>() {

        @Override
        public void apply(Object msg) throws Exception {
            if(msg instanceof DatabaseTableInfo) {
                Map<String, DatabaseColumnInfo> columnInfos = 
                    Stream.of(((DatabaseTableInfo) msg).getColumns())
                        .collect(Collectors.toMap(
                            columnInfo -> columnInfo.getName(),
                            columnInfo -> columnInfo));

                List<String> columnNames = originalMsg.getColumnNames();
                if(columnNames.stream().allMatch(columnInfos::containsKey)) {
                    log.debug("fetching records");
                    transaction.tell(
                        new FetchTable(
                            tableName, 
                            columnNames.stream()
                                .map(columnInfos::get)
                                .collect(Collectors.toList()),
                            originalMsg.getMessageSize()),
                        getSelf());

                    getContext().become(onReceiveItems());
                } else {
                    log.error("missing column(s)");
                    unavailable();
                }
            } else if(msg instanceof ReceiveTimeout) {
                log.debug("timeout received");
                unavailable();
            } else {
                unhandled(msg);
            }
        }

    };
}

private Procedure<Object> onReceiveItemRecords() {
    return new Procedure<Object>() {

        @Override
        public void apply(Object msg) throws Exception {
            if(msg instanceof SDEItemInfo) {
                log.debug("item info received");

                SDEItemInfo itemInfo = (SDEItemInfo)msg;
                SDEItemInfoType type = itemInfo.getType();
                String tableName = itemInfo.getPhysicalname();

                if(SDEItemInfoType.FEATURE_CLASS == type ||
                    SDEItemInfoType.TABLE == type) {
                    log.debug("tableName: {}", tableName);

                    SDEGetVectorDatasetHandler.this.tableName = tableName;
                    transaction.tell(new DescribeTable(tableName), getSelf());
                    getContext().become(onReceiveDatabaseTableInfo());
                } else {
                    log.error("wrong item type: {}", type);
                    unavailable();
                }
            } else if(msg instanceof ReceiveTimeout) {
                log.debug("timeout received");
                unavailable();
            }
        }

    };
}

@Override
public void onReceive(Object msg) throws Exception {
    if(msg instanceof TransactionCreated) {
        log.debug("transaction created");

        transaction = ((TransactionCreated)msg).getActor();
        ActorRef recordsReceiver = getContext().actorOf(
            SDEReceiveSingleItemInfo.props(getSelf()), 
            "item-records-receiver");

        try {
            databaseScheme = databaseConfig.getString("scheme");
        } catch(ConfigException.Missing cem) {
            databaseScheme = "SDE";
        }

        log.debug("database scheme before calling get fetch table: " + databaseScheme);

        transaction.tell(
            SDEUtils.getFetchTable(SDEUtils.getItemsFilter(originalMsg.getIdentification()), databaseScheme),
            recordsReceiver);
        getContext().become(onReceiveItemRecords());
    } else if(msg instanceof ReceiveTimeout) {
        log.debug("timeout received", msg);
        unavailable();
    } else {
        unhandled(msg);
    }
}

private 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());
            }
        }
    };
}

@Override
public void onReceive(Object msg) throws Exception {
    if(msg instanceof ReceiveTimeout) {
        log.debug("timeout");           
        getContext().parent().tell(new StopPackager(targetName), getSelf());
    } else if(msg instanceof SetPersistent) {
        log.debug("set persistent");
        getContext().parent().tell(msg, getSelf());
    } else {
        final String sourceName;

        ActorRef sender = getSender();
        if(sender.equals(getContext().system().deadLetters())) {
            log.debug("no sender");         
            sourceName = null;
        } else {            
            log.debug("sender: " + sender);
            String sourcePath = sender.path().toString();
            if(sourcePath.startsWith(pathPrefix)) {
                sourceName = sourcePath.substring(pathPrefix.length() + 1);
            } else {
                log.error("sourcePath: " + sourcePath + " pathPrefix: " + pathPrefix + " msg: " + msg.toString());
                throw new IllegalStateException("sender is not a child of container actor");
            }
        }

        messageTarget.tell(new Envelope(targetName, msg, sourceName), getSelf());
    }
}

@Override
public void onReceive(Object msg) throws Exception {
    if(msg instanceof ReceiveTimeout) {
        log.debug("timeout");

        promise.failure(new TimeoutException("ask timeout: " + timeout.toString()));
    } else {
        log.debug("answer received");

        promise.success(new AskResponse<>(msg, getSender()));
    }

    getContext().stop(self());
}

@Override
public void onReceive(Object msg) throws Exception {
    if(msg instanceof byte[]) {
        handleFileChunk((byte[])msg);           
    } else if(msg instanceof End) {
        handleEnd();            
    } else if(msg instanceof ReceiveTimeout) {
        log.error("timeout");
        getContext().stop(getSelf());
    } else {
        unhandled(msg);
    }
}

private Procedure<Object> afterDelete(ActorRef sender, Class<?> entity, Object beforeResult) {
    return new Procedure<Object>() {

        @Override
        @SuppressWarnings("unchecked")
        public void apply(Object msg) throws Exception {
            if(msg instanceof ReceiveTimeout) {
                log.error("timeout while waiting for event to complete");

                finish();
            } else if(msg instanceof EventCompleted && getSender().equals(sender)) {
                log.debug("event completed");

                onDeleteAfter.get(entity).accept(beforeResult, ((EventCompleted<?>)msg).getValue());

                finish();
            } else if(msg instanceof EventFailed && getSender().equals(sender)) {
                log.debug("event failed");

                finish();
            } else {
                log.debug("message stashed: {} from {}", msg, getSender());

                stash();
            }
        }

        private void finish() {
            log.debug("delete event finished");

            unstashAll();
            getContext().setReceiveTimeout(Duration.Inf());
            getContext().become(receive());
        }

    };
}

@Override
public void onReceive(Object msg) throws Exception {
    if(msg instanceof ReceiveTimeout) {
        log.error("timeout while waiting for listeners");

        handler.tell(origMsg, origSender);
        getContext().stop(getSelf());
    } else if(msg instanceof EventWaiting || msg instanceof EventFailed) {
        ActorRef listener = getSender();

        log.debug("listener ready: {}", listener);

        pendingListeners.remove(listener);

        if(msg instanceof EventFailed) {
            listeners.remove(listener);
        }

        if(pendingListeners.isEmpty()) {
            log.debug("listeners completed before function");

            handler.tell(origMsg, getSelf());
            getContext().become(waitingForAnswer());
        }
    } else {
        unhandled(msg);
    }
}

protected final void onReceiveCommon(Object msg) {
    if(msg instanceof ReceiveTimeout) {             
        handleTimeout();
    } else if(msg instanceof FinalizeSession) {
        handleFinalizeSession((FinalizeSession)msg);
    } else if(msg instanceof Failure) {
        handleFailure((Failure)msg);
    } else if(msg instanceof End) {
        handleEnd((End)msg);
    } else {
        onReceiveElse(msg);
    }
}

private Procedure<Object> importing() {
    return new Procedure<Object>() {

        private long lastSeq = -1;

        private int retries = maxRetries;

        @Override
        public void apply(Object msg) throws Exception {
            if(msg instanceof Item<?>) {
                Item<?> item = (Item<?>)msg;
                long seq = item.getSequenceNumber();
                if(seq == lastSeq + 1) {
                    lastSeq = seq;
                    lastItemSender = getSender();
                    retries = maxRetries;

                    handleItemContent(item.getContent());
                } else {
                    log.warning("unexpected sequence number: {}, expected: {}", seq, lastSeq + 1);
                }
            } else if(msg instanceof ReceiveTimeout && retries > 0 && lastItemSender != null) {
                log.warning("timemout, requesting retry");
                lastItemSender.tell(new NextItem(lastSeq + 1), getSelf());
                retries--;
            } else {
                onReceiveCommon(msg);
            }
        }           
    };
}

@Override
@SuppressWarnings("unchecked")
public void onReceive(Object msg) throws Exception {
    if(msg instanceof ReceiveTimeout) {
        handleTimeout();
    } else if(msg instanceof Item) {
        handleDataset(((Item<Dataset>)msg).getContent());
    } else if(msg instanceof End) {
        handleEnd();
    } else if(msg instanceof Unavailable) {
        handleUnavailable();
    } else {
        unhandled(msg);
    }
}

@Override
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");

        for(Attachment attachment : ((DatasetInfo) msg).getAttachments()) {
            if(attachment.getAttachmentType().equals(AttachmentType.METADATA)) {
                log.debug("metadata document found");

                MetadataDocumentFactory metadataDocumentFactory = new MetadataDocumentFactory();
                MetadataDocument metadataDocument = metadataDocumentFactory.parseDocument((byte[])attachment.getContent());

                sender.tell(metadataDocument, getSelf());

                break;
            }
        }

        getContext().stop(getSelf());
    } else {
        unhandled(msg);
    }
}