mseemann

// SPDX-License-Identifier: MIT
pragma solidity >=0.4.22 <0.9.0;

contract SampleContract {
  uint storedData;

  function set(uint x) public {
    storedData = x;
  }

mseemann

max_over_time(sync_last_processed_doc_at[24h]) - max_over_time(sync_last_seen_doc_at[24h])

mseemann

max_over_time(sync_last_seen_oplog_at[24h]) - ignoring(instance, job)  max_over_time(sync_last_seen_doc_at[24h])

mseemann

   Gauge.builder("sync.last.seen.doc.at", this, listener -> listener.lastSeenAt.get())
            .strongReference(true)
            .register(Metrics.globalRegistry);
    Gauge.builder("sync.last.processed.doc.at", this, listener -> listener.lastProcessedAt.get())
            .strongReference(true)
            .register(Metrics.globalRegistry);

mseemann

  Gauge.builder("sync.last.seen.oplog.at", this, mongoStats -> mongoStats.lastOplogAt.get())
          .strongReference(true)
          .register(Metrics.globalRegistry);

mseemann

Query tailQ = new Query().addCriteria(Criteria.where("ns").is("streams.user"));
reactiveMongoTemplate.tail(tailQ, OplogEntry.class, "oplog.rs")
        .retryWhen(Retry.indefinitely().doAfterRetryAsync(signal -> Mono.delay(Duration.ofSeconds(10)).then()))
        .subscribe(entry -> {
            log.info("tailed entry: {}", entry);
            lastOplogAt.set(entry.ts().getTime() * 1000L);
        });

mseemann

@Service
@Slf4j
@Transactional(propagation = Propagation.REQUIRED)
@AllArgsConstructor
public class SyncToPostgresService {

    UserEntityRepository userEntityRepository;
    ResumeTokenRepository resumeTokenRepository;

    public void syncInASingleTx(User document, String bookmarkToken, Long tokenTimeStamp) {

mseemann

reactiveMongoTemplate
        .changeStream(User.class)
        .watchCollection("user")
        .listen()
        .subscribe(event -> {
            log.info("change stream event {}", event);
            BsonDocument newResumeToken = Objects.requireNonNull(event.getRaw()).getResumeToken().asDocument();
            BsonString token = newResumeToken.getString("_data");
            var timeStampInMs = Objects.requireNonNull(event.getBsonTimestamp()).getTime() * 1000L;

mseemann

spring:
  data:
    mongodb:
      host:                                             localhost
      port:                                             27017
      database:                                         streams

mseemann

@RestController
@AllArgsConstructor
public class UserController {

    UserRepository userRepository;

    @GetMapping("/user")
    public Flux<User> getAllUser() {
        return userRepository.findAll();