<commit_before>package com.example.person;

import com.example.ip.IpService;
import com.mongodb.util.JSON;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import reactor.core.publisher.BufferOverflowStrategy;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.List;

@Service
public class PersonService {
    private final Logger log = LoggerFactory.getLogger(PersonRouter.class);

    private final IpService ipService;

    private final PersonRepository repository;

    private final PersonSender sender;

    public PersonService(IpService ipService, PersonRepository repository, PersonSender sender) {
        this.ipService = ipService;
        this.repository = repository;
        this.sender = sender;
    }

    public Mono<Person> addPerson(Mono<Person> mono) {
        return mono.flatMap(this::addIpInfo)
                .flatMap(repository::save)
                .flatMap(sender::send);
    }

    private Mono<Person> addIpInfo(Person person) {
        return ipService.getIpInfo(person.getIp()).map(ipInfo -> person.copyWithIpInfo(JSON.parse(ipInfo)));
    }

    public void updateIpInfo() {
        repository.findAll()
                .buffer(10)
                .onBackpressureBuffer(1000, BufferOverflowStrategy.ERROR)
                .retry(2)
                .parallel(2)
                .flatMap(this::updateIpInfo)
                .subscribe(
                        p -> log.trace("Updated IP information for person", p),
                        t -> log.error("Failed IP information update stream", t)
                );
    }

    private Flux<Person> updateIpInfo(List<Person> batch) {
        return batch.stream()
                .map(p -> repository.save(addIpInfo(p)))
                .reduce(Flux.empty(), (Flux<Person> previous, Flux<Person> next) -> previous.concatWith(next));
    }
}<commit_msg>Fixed stream log output<commit_after>package com.example.person;

import com.example.ip.IpService;
import com.mongodb.util.JSON;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import reactor.core.publisher.BufferOverflowStrategy;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.List;

@Service
public class PersonService {
    private final Logger log = LoggerFactory.getLogger(PersonRouter.class);

    private final IpService ipService;

    private final PersonRepository repository;

    private final PersonSender sender;

    public PersonService(IpService ipService, PersonRepository repository, PersonSender sender) {
        this.ipService = ipService;
        this.repository = repository;
        this.sender = sender;
    }

    public Mono<Person> addPerson(Mono<Person> mono) {
        return mono.flatMap(this::addIpInfo)
                .flatMap(repository::save)
                .flatMap(sender::send);
    }

    private Mono<Person> addIpInfo(Person person) {
        return ipService.getIpInfo(person.getIp()).map(ipInfo -> person.copyWithIpInfo(JSON.parse(ipInfo)));
    }

    public void updateIpInfo() {
        repository.findAll()
                .buffer(10)
                .onBackpressureBuffer(1000, BufferOverflowStrategy.ERROR)
                .retry(2)
                .parallel(2)
                .flatMap(this::updateIpInfo)
                .subscribe(
                        p -> log.info("Updated IP information for person with id {}", p.getId()),
                        t -> log.error("Failed IP information update stream", t)
                );
    }

    private Flux<Person> updateIpInfo(List<Person> batch) {
        return batch.stream()
                .map(p -> repository.save(addIpInfo(p)))
                .reduce(Flux.empty(), (Flux<Person> previous, Flux<Person> next) -> previous.concatWith(next));
    }
}
{'fraction_non_alphanumeric': 0.10524961399897066, 'fraction_numerical': 0.0051466803911477095, 'mean_word_length': 2.599074074074074, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 27, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6266192', 'n_tokens_mistral': 1108, 'n_tokens_neox': 1046, 'n_words': 221}