@Override
    public ChannelOutput receive() throws IOException {
      Response response;
      if (metadataRequest != null) {
        List<ReplicaMetadataResponseInfo> responseInfoList = new ArrayList<>();
        for (ReplicaMetadataRequestInfo requestInfo : metadataRequest.getReplicaMetadataRequestInfoList()) {
          List<MessageInfo> messageInfosToReturn = new ArrayList<>();
          List<MessageInfo> allMessageInfos =
              host.infosByPartition.getOrDefault(requestInfo.getPartitionId(), new ArrayList<>());
          int startIndex = ((MockFindToken) (requestInfo.getToken())).getIndex();
          int endIndex = Math.min(allMessageInfos.size(), startIndex + maxSizeToReturn);
          Set<StoreKey> processedKeys = new HashSet<>();
          for (int i = startIndex; i < endIndex; i++) {
            StoreKey key = allMessageInfos.get(i).getStoreKey();
            if (processedKeys.add(key)) {
              messageInfosToReturn.add(getMergedMessageInfo(key, allMessageInfos));
            }
          }
          long bytesRead = allMessageInfos.size() > 0 ? allMessageInfos.subList(0, endIndex)
              .stream()
              .mapToLong(MessageInfo::getSize)
              .sum() : 0;
          long total = allMessageInfos.stream().mapToLong(MessageInfo::getSize).sum();
          ReplicaMetadataResponseInfo replicaMetadataResponseInfo =
              new ReplicaMetadataResponseInfo(requestInfo.getPartitionId(), requestInfo.getReplicaType(),
                  new MockFindToken(endIndex, bytesRead), messageInfosToReturn, total - bytesRead,
                  ReplicaMetadataResponse.getCompatibleResponseVersion(metadataRequest.getVersionId()));
          responseInfoList.add(replicaMetadataResponseInfo);
        }
        response = new ReplicaMetadataResponse(1, "replicametadata", ServerErrorCode.NoError, responseInfoList,
            ReplicaMetadataResponse.getCompatibleResponseVersion(metadataRequest.getVersionId()));
        metadataRequest = null;
      } else {
        List<PartitionResponseInfo> responseInfoList = new ArrayList<>();
        for (PartitionRequestInfo requestInfo : getRequest.getPartitionInfoList()) {
          List<MessageInfo> infosForPartition = infosToReturn.get(requestInfo.getPartition());
          PartitionResponseInfo partitionResponseInfo;
          if (!getRequest.getGetOption().equals(GetOption.Include_All) && !getRequest.getGetOption()
              .equals(GetOption.Include_Deleted_Blobs) && infosForPartition.stream().anyMatch(MessageInfo::isDeleted)) {
            partitionResponseInfo = new PartitionResponseInfo(requestInfo.getPartition(), ServerErrorCode.BlobDeleted);
          } else if (!getRequest.getGetOption().equals(GetOption.Include_All) && !getRequest.getGetOption()
              .equals(GetOption.Include_Expired_Blobs) && infosForPartition.stream().anyMatch(MessageInfo::isExpired)) {
            partitionResponseInfo = new PartitionResponseInfo(requestInfo.getPartition(), ServerErrorCode.BlobExpired);
          } else {
            partitionResponseInfo =
                new PartitionResponseInfo(requestInfo.getPartition(), infosToReturn.get(requestInfo.getPartition()),
                    messageMetadatasToReturn.get(requestInfo.getPartition()));
          }
          responseInfoList.add(partitionResponseInfo);
        }
        response = new GetResponse(1, "replication", responseInfoList, new MockSend(buffersToReturn),
            ServerErrorCode.NoError);
        getRequest = null;
      }
      ByteBuffer buffer = ByteBuffer.allocate((int) response.sizeInBytes());
      ByteBufferOutputStream stream = new ByteBufferOutputStream(buffer);
      WritableByteChannel channel = Channels.newChannel(stream);
      while (!response.isSendComplete()) {
        response.writeTo(channel);
      }
      buffer.flip();
      buffer.getLong();
      response.release();
      return new ChannelOutput(new DataInputStream(new ByteBufferInputStream(buffer)), buffer.remaining());
    }
Implementation of {@link ConnectedChannel} that fetches message infos or blobs based on the type of request.
ambry-replication/src/test/java/com/github/ambry/replication/MockConnectionPool.java