Монго реактивное сохранение: запись не сохраняется / обновляется при каждом repo.save (), появляется только после завершения процесса

Question

Я работаю над проблемой рюкзака и пытаюсь решить ее с помощью пружинного реактива с mongo db.Идея состоит в том, что когда я получил входные данные о проблеме ранца от пользователя, я установил статус задачи на «ОТПРАВЛЕНО», а затем вернул ответ пользователю и начал параллельно работать над решением, чтобы найти решение этой проблемы ранца.Во время этого процесса состояние задачи / задачи будет изменено с отправлено-> запущено-> завершено.На каждом этапе статус должен быть сохранен в базе данных.так что эту задачу можно отслеживать в любой момент времени.Проблема с приведенным ниже кодом заключается в том, что он просто вставляет запись в базу данных, когда решение проблемы готово, это означает, что статус задачи завершен.Я хочу, чтобы запись вставлялась с начальным статусом SUBMITTED и позже во время обработки обновлялась последовательно со статусами «STARTED» и «COMPLETED».К вашему сведению: я новичок в весенних реактивных вещах

, и я хотел бы знать, как я могу отправить немедленное подтверждение со статусом "ОТПРАВЛЕНО".Обновленный код до данных можно найти в моем репозитории github

https://github.com/Manishmbm2010/Knapsack-reactive

Сервисный код, где я сталкиваюсь с проблемой

public void simpleSave(Knapsack knapsack) {
    knapsackReactiveRepository
        .save(knapsack)
        .single()
        .subscribe(
            x -> System.out.println(x.getTaskId() + " task status is " + x.getStatus()),
            Throwable::printStackTrace,
            () -> calculateSolutionForKnapsack(knapsack));
  }

  private void calculateSolutionForKnapsack(Knapsack knapsack) {
    knapsack.getTimestamps().setStarted(Instant.now().getEpochSecond());
    knapsack.setStatus(Status.STARTED);
    knapsackReactiveRepository
        .save(knapsack)
        .delayElement(Duration.ofSeconds(10))
        .subscribeOn(Schedulers.fromExecutor(executor))
        .subscribe(
            x -> System.out.println(x.getTaskId() + " task status is " + x.getStatus()),
            Throwable::printStackTrace,
            () -> {
              Problem problem = knapsack.getProblem();
              List<Integer> itemsToPutInKnapsack =
                  knapsackAlgo(
                      problem.getCapacity(),
                      problem.getWeights(),
                      problem.getValues(),
                      problem.getValues().length);
              knapsack.getSolution().setItems(itemsToPutInKnapsack);
              knapsack.getTimestamps().setCompleted(Instant.now().getEpochSecond());
              knapsack.setStatus(Status.COMPLETED);
              knapsack
                  .getSolution()
                  .setTime(timeTakenToSolveTheKnapsackProblem(knapsack.getTimestamps()));
              knapsackReactiveRepository
                  .save(knapsack)
                  .delayElement(Duration.ofSeconds(10))
                  .subscribe(
                      x -> System.out.println(x.getTaskId() + " task status is " + x.getStatus()),
                      Throwable::printStackTrace,
                      () -> System.out.println("Processing completed"));
            });
  }

Рюкзак Модель

import lombok.Data;
import org.springframework.data.annotation.Id;
import org.springframework.data.mongodb.core.mapping.Document;


@Document
@Data
// @JsonView("knapsack")
public class Knapsack {

  @Id private String taskId;;
  private Problem problem;
  private Solution solution;
  private TimeStamp timestamps;
  private Status status;

  public Knapsack() {
    this.timestamps = new TimeStamp();
    solution = new Solution();
    this.status = Status.SUBMITTED;
  }

  public enum Status {
    SUBMITTED,
    STARTED,
    COMPLETED
  }
}

Контроллер

@RequestMapping(path = "/knapsack", method = RequestMethod.POST)
  public Mono<Knapsack> submitKnapsackProblem(@RequestBody Knapsack knapsack) {
    return knapsackService.saveKnapsackProblem(knapsack);
  }

вход

{
  "problem":{
    "capacity":10,
  "weights":[5,4,3],
  "values":[100,50,200]
  }
}

Ожидаемый вывод в качестве подтверждения:

{
{
    "taskId": "5cf2f4ec05a36459cd6faab7",
    "problem": {
        "capacity": 10,
        "weights": [
            5,
            4,
            3
        ],
        "values": [
            100,
            50,
            200
        ]
    },
    "solution": {
        "items": null,
        "time": 0
    },
    "timestamps": {
        "submitted": 1559426284,
        "started": null,
        "completed": null
    },
    "status": "SUBMITTED"
}

FinalOutput, когда готовое решение для рюкзака готово для извлечениядругой конечной точкой:

{
    "taskId": "5cf2f4ec05a36459cd6faab7",
    "problem": {
        "capacity": 10,
        "weights": [
            5,
            4,
            3
        ],
        "values": [
            100,
            50,
            200
        ]
    },
    "solution": {
        "items": [
            0,
            2
        ],
        "time": 1
    },
    "timestamps": {
        "submitted": 1559426284,
        "started": 1559426285,
        "completed": 1559426285
    },
    "status": "COMPLETED"
}