Source code

001/*
002 * Copyright 2017-2022 original authors
003 *
004 * Licensed under the Apache License, Version 2.0 (the "License");
005 * you may not use this file except in compliance with the License.
006 * You may obtain a copy of the License at
007 *
008 * https://www.apache.org/licenses/LICENSE-2.0
009 *
010 * Unless required by applicable law or agreed to in writing, software
011 * distributed under the License is distributed on an "AS IS" BASIS,
012 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
013 * See the License for the specific language governing permissions and
014 * limitations under the License.
015 */
016package io.micronaut.maven.services;
017
018import com.github.dockerjava.api.DockerClient;
019import com.github.dockerjava.api.async.ResultCallback;
020import com.github.dockerjava.api.command.BuildImageCmd;
021import com.github.dockerjava.api.command.BuildImageResultCallback;
022import com.github.dockerjava.api.command.CreateContainerCmd;
023import com.github.dockerjava.api.command.CreateContainerResponse;
024import com.github.dockerjava.api.command.ExecCreateCmd;
025import com.github.dockerjava.api.command.InspectImageCmd;
026import com.github.dockerjava.api.command.InspectImageResponse;
027import com.github.dockerjava.api.command.KillContainerCmd;
028import com.github.dockerjava.api.command.PushImageCmd;
029import com.github.dockerjava.api.command.RemoveContainerCmd;
030import com.github.dockerjava.api.command.RemoveImageCmd;
031import com.github.dockerjava.api.command.StartContainerCmd;
032import com.github.dockerjava.api.command.WaitContainerCmd;
033import com.github.dockerjava.api.command.WaitContainerResultCallback;
034import com.github.dockerjava.api.exception.DockerClientException;
035import com.github.dockerjava.api.exception.ConflictException;
036import com.github.dockerjava.api.exception.DockerException;
037import com.github.dockerjava.api.exception.NotFoundException;
038import com.github.dockerjava.api.model.AuthConfig;
039import com.github.dockerjava.api.model.AuthConfigurations;
040import com.github.dockerjava.api.model.AuthResponse;
041import com.github.dockerjava.api.model.Bind;
042import com.github.dockerjava.api.model.BuildResponseItem;
043import com.github.dockerjava.api.model.Frame;
044import com.github.dockerjava.api.model.HostConfig;
045import com.github.dockerjava.api.model.Info;
046import com.github.dockerjava.core.DefaultDockerClientConfig;
047import com.github.dockerjava.core.DockerClientConfig;
048import com.github.dockerjava.core.DockerClientImpl;
049import com.github.dockerjava.zerodep.ZerodepDockerHttpClient;
050import com.google.cloud.tools.jib.api.Credential;
051import io.micronaut.maven.DockerfileMojo;
052import io.micronaut.core.annotation.Internal;
053import io.micronaut.maven.jib.JibConfigurationService;
054import org.apache.commons.compress.archivers.tar.TarArchiveEntry;
055import org.apache.commons.compress.archivers.tar.TarArchiveInputStream;
056import org.apache.commons.io.FileUtils;
057import org.apache.commons.io.IOUtils;
058import org.apache.commons.lang3.StringUtils;
059import org.apache.maven.project.MavenProject;
060import org.slf4j.Logger;
061import org.slf4j.LoggerFactory;
062import org.testcontainers.containers.output.FrameConsumerResultCallback;
063import org.testcontainers.containers.output.OutputFrame;
064import org.testcontainers.containers.output.Slf4jLogConsumer;
065import org.testcontainers.utility.DockerImageName;
066import org.testcontainers.utility.RegistryAuthLocator;
067
068import javax.inject.Inject;
069import javax.inject.Singleton;
070import java.io.File;
071import java.io.IOException;
072import java.io.InputStream;
073import java.nio.charset.StandardCharsets;
074import java.nio.file.Files;
075import java.nio.file.Path;
076import java.nio.file.StandardCopyOption;
077import java.util.ArrayList;
078import java.util.List;
079import java.util.Map;
080import java.util.Optional;
081import java.util.concurrent.TimeUnit;
082import java.util.function.Consumer;
083
084/**
085 * Provides methods to work with Docker images.
086 *
087 * @author Álvaro Sánchez-Mariscal
088 * @since 1.1
089 */
090@Singleton
091public class DockerService {
092
093    private static final Logger LOG = LoggerFactory.getLogger(DockerService.class);
094
095    private final DockerClientConfig config;
096    private final MavenProject mavenProject;
097    private final JibConfigurationService jibConfigurationService;
098    private DockerClient dockerClient;
099
100    @SuppressWarnings("CdiInjectionPointsInspection")
101    @Inject
102    public DockerService(MavenProject mavenProject, JibConfigurationService jibConfigurationService) {
103        this.mavenProject = mavenProject;
104        this.jibConfigurationService = jibConfigurationService;
105        this.config = DefaultDockerClientConfig.createDefaultConfigBuilder().build();
106    }
107
108    /**
109     * Uses the given Docker client instead of connecting to the configured Docker host. For tests.
110     *
111     * @param mavenProject the Maven project
112     * @param jibConfigurationService the Jib configuration service
113     * @param dockerClient the Docker client
114     */
115    DockerService(MavenProject mavenProject, JibConfigurationService jibConfigurationService, DockerClient dockerClient) {
116        this(mavenProject, jibConfigurationService);
117        this.dockerClient = dockerClient;
118    }
119
120    private DockerClient getDockerClient() {
121        if (dockerClient == null) {
122            var httpClient = new ZerodepDockerHttpClient.Builder()
123                    .dockerHost(config.getDockerHost())
124                    .sslConfig(config.getSSLConfig())
125                    .build();
126            dockerClient = DockerClientImpl.getInstance(config, httpClient);
127        }
128        return dockerClient;
129    }
130
131    /**
132     * @param dockerfileName the name of the Dockerfile to load
133     * @return the {@link BuildImageCmd} by loading the given Dockerfile as classpath resource.
134     */
135    public BuildImageCmd buildImageCmd(String dockerfileName) throws IOException {
136        verifyDockerRunning();
137        BuildImageCmd buildImageCmd = getDockerClient().buildImageCmd(loadDockerfileAsResource(dockerfileName));
138        maybeConfigureBuildAuth(buildImageCmd);
139        return buildImageCmd;
140    }
141
142    private void maybeConfigureBuildAuth(BuildImageCmd buildImageCmd) {
143        jibConfigurationService.getFromImage().ifPresent(image -> {
144            Optional<Credential> fromCredentials = jibConfigurationService.getFromCredentials();
145            Optional<Credential> credential = fromCredentials.or(() -> jibConfigurationService.resolveCredentialForImage(image, LOG));
146            credential.ifPresent(cred -> {
147                var username = cred.getUsername();
148                var password = cred.getPassword();
149                AuthConfig authConfig = getAuthConfigFor(image, username, password);
150                var authConfigurations = new AuthConfigurations();
151                authConfigurations.addConfig(authConfig);
152                buildImageCmd.withBuildAuthConfigs(authConfigurations);
153            });
154        });
155    }
156
157    /**
158     * @return a default {@link BuildImageCmd}.
159     */
160    public BuildImageCmd buildImageCmd() {
161        verifyDockerRunning();
162        BuildImageCmd buildImageCmd = getDockerClient().buildImageCmd();
163        maybeConfigureBuildAuth(buildImageCmd);
164        return buildImageCmd;
165    }
166
167    /**
168     * Builds the Docker image from the given {@link BuildImageCmd} builder.
169     *
170     * @param builder The builder to use.
171     * @return The resulting image ID.
172     */
173    public String buildImage(BuildImageCmd builder) {
174        verifyDockerRunning();
175        if (builder.getBuildArgs() != null) {
176            builder.getBuildArgs().forEach((k, v) -> LOG.info("Using {}: {}", k, v));
177        }
178        BuildImageResultCallback resultCallback = new BuildImageResultCallback() {
179            @Override
180            public void onNext(BuildResponseItem item) {
181                super.onNext(item);
182                if (item.isErrorIndicated() && item.getErrorDetail() != null) {
183                    LOG.error(item.getErrorDetail().getMessage());
184                } else if (item.getStream() != null) {
185                    String msg = StringUtils.removeEnd(item.getStream(), System.lineSeparator());
186                    LOG.info(msg);
187                }
188            }
189        };
190
191        return builder
192            .exec(resultCallback)
193            .awaitImageId();
194    }
195
196    /**
197     * Creates a container based on a given image, and runs it.
198     *
199     * @param imageId the image to use
200     * @param timeoutSeconds the timeout in seconds for the container to finish execution
201     * @param checkpointNetworkName the name of the network to use for the container
202     * @param binds the bind mounts to use
203     */
204    public void runPrivilegedImageAndWait(String imageId, Integer timeoutSeconds, String checkpointNetworkName, String... binds) throws IOException {
205        String containerId = createContainer(imageId, checkpointNetworkName, true, Map.of(), binds);
206        startAndWait(containerId, imageId, timeoutSeconds);
207    }
208
209    /**
210     * Creates an unprivileged container from the given image, without starting it.
211     *
212     * @param imageId the image to use
213     * @param networkName the name of the network to use for the container, if any
214     * @param environment the environment variables to set, on top of the image ones
215     * @return the container ID
216     * @since 5.1.0
217     */
218    @Internal
219    public String createContainer(String imageId, String networkName, Map<String, String> environment) {
220        return createContainer(imageId, networkName, false, environment);
221    }
222
223    private String createContainer(String imageId, String networkName, boolean privileged, Map<String, String> environment,
224                                   String... binds) {
225        verifyDockerRunning();
226        try (CreateContainerCmd create = getDockerClient().createContainerCmd(imageId)) {
227            HostConfig hostConfig = create.getHostConfig();
228            if (hostConfig == null) {
229                throw new DockerClientException("When setting binds and privileged, hostConfig was null.  Please check your docker installation and try again");
230            }
231            if (privileged) {
232                hostConfig.withPrivileged(true);
233            }
234            if (networkName != null) {
235                hostConfig.withNetworkMode(networkName);
236            }
237            for (String bind : binds) {
238                hostConfig.withBinds(Bind.parse(bind));
239            }
240            if (!environment.isEmpty()) {
241                create.withEnv(environment.entrySet().stream().map(e -> e.getKey() + "=" + e.getValue()).toList());
242            }
243            return create.exec().getId();
244        }
245    }
246
247    /**
248     * Starts a container and waits for it to exit with status 0. Otherwise, logs its output and fails.
249     *
250     * @param containerId the container
251     * @param imageId the image of the container, for the error message
252     * @param timeoutSeconds the timeout in seconds for the container to finish execution
253     * @throws IOException if the container does not exit with status 0 within the timeout
254     * @since 5.1.0
255     */
256    @Internal
257    public void startAndWait(String containerId, String imageId, Integer timeoutSeconds) throws IOException {
258        startContainer(containerId);
259        LOG.info("Waiting {} seconds for completion", timeoutSeconds);
260        int exitCode;
261        try {
262            exitCode = awaitExit(containerId, timeoutSeconds);
263        } catch (IOException e) {
264            logContainerOutput(containerId);
265            throw e;
266        }
267        if (exitCode != 0) {
268            logContainerOutput(containerId);
269            throw new IOException("Image " + imageId + " exited with code " + exitCode);
270        }
271    }
272
273    /**
274     * Starts a container.
275     *
276     * @param containerId the container
277     * @since 5.1.0
278     */
279    @Internal
280    public void startContainer(String containerId) {
281        try (StartContainerCmd start = getDockerClient().startContainerCmd(containerId)) {
282            start.exec();
283            LOG.info("Container started: {}", containerId);
284        }
285    }
286
287    /**
288     * Waits for a container to exit.
289     *
290     * @param containerId the container
291     * @param timeoutSeconds the timeout in seconds
292     * @return the exit code of the container
293     * @throws IOException if the container does not exit within the timeout
294     * @since 5.1.0
295     */
296    @Internal
297    public int awaitExit(String containerId, int timeoutSeconds) throws IOException {
298        try (WaitContainerCmd wait = getDockerClient().waitContainerCmd(containerId)) {
299            WaitContainerResultCallback waitResult = wait.start();
300            try {
301                return waitResult.awaitStatusCode(timeoutSeconds, TimeUnit.SECONDS);
302            } catch (DockerClientException e) {
303                throw new IOException("Container " + containerId + " did not exit within " + timeoutSeconds + " seconds", e);
304            }
305        }
306    }
307
308    /**
309     * Sends a signal to the main process of a running container.
310     *
311     * @param containerId the container
312     * @param signal the signal, for example {@code SIGTERM}
313     * @since 5.1.0
314     */
315    @Internal
316    public void signalContainer(String containerId, String signal) {
317        try (KillContainerCmd kill = getDockerClient().killContainerCmd(containerId)) {
318            kill.withSignal(signal).exec();
319        } catch (ConflictException | NotFoundException e) {
320            LOG.debug("Container {} is not running: {}", containerId, e.getMessage());
321        }
322    }
323
324    /**
325     * Runs a command in a running container and waits for it.
326     *
327     * @param containerId the container
328     * @param timeoutSeconds the timeout in seconds
329     * @param output receives the output of the command, line by line
330     * @param command the command and its arguments
331     * @return the exit code of the command
332     * @throws IOException if the command does not finish within the timeout
333     * @since 5.1.0
334     */
335    @Internal
336    public int execInContainer(String containerId, int timeoutSeconds, Consumer<String> output, String... command) throws IOException {
337        String execId;
338        try (ExecCreateCmd create = getDockerClient().execCreateCmd(containerId)) {
339            execId = create.withCmd(command).withAttachStdout(true).withAttachStderr(true).exec().getId();
340        }
341        try (var callback = new LineCallback(output)) {
342            boolean completed = getDockerClient().execStartCmd(execId).exec(callback).awaitCompletion(timeoutSeconds, TimeUnit.SECONDS);
343            callback.flush();
344            if (!completed) {
345                throw new IOException("Command " + command[0] + " did not finish within " + timeoutSeconds + " seconds");
346            }
347        } catch (InterruptedException e) {
348            Thread.currentThread().interrupt();
349            throw new IOException("Interrupted while running " + command[0], e);
350        }
351        Long exitCode = getDockerClient().inspectExecCmd(execId).exec().getExitCodeLong();
352        return exitCode == null ? -1 : exitCode.intValue();
353    }
354
355    /**
356     * Runs an image with the given entrypoint, waits for it and returns its output.
357     *
358     * @param imageId the image to use
359     * @param timeoutSeconds the timeout in seconds
360     * @param entrypoint the entrypoint, which replaces the image entrypoint and command
361     * @return the exit code and the output of the container
362     * @throws IOException if the container does not exit within the timeout
363     * @since 5.1.0
364     */
365    @Internal
366    public ContainerOutput runAndCaptureOutput(String imageId, int timeoutSeconds, List<String> entrypoint) throws IOException {
367        verifyDockerRunning();
368        String containerId;
369        try (CreateContainerCmd create = getDockerClient().createContainerCmd(imageId)) {
370            containerId = create.withEntrypoint(entrypoint).exec().getId();
371        }
372        try {
373            startContainer(containerId);
374            int exitCode = awaitExit(containerId, timeoutSeconds);
375            var lines = new ArrayList<String>();
376            try (var callback = new LineCallback(lines::add)) {
377                getDockerClient().logContainerCmd(containerId).withStdOut(true).withStdErr(true).exec(callback).awaitCompletion();
378                callback.flush();
379            } catch (InterruptedException _) {
380                Thread.currentThread().interrupt();
381            }
382            return new ContainerOutput(exitCode, String.join("\n", lines));
383        } finally {
384            removeContainer(containerId);
385        }
386    }
387
388    /**
389     * Logs the output of a container.
390     *
391     * @param containerId the container
392     * @since 5.1.0
393     */
394    @Internal
395    public void logContainerOutput(String containerId) {
396        final Slf4jLogConsumer stdoutConsumer = new Slf4jLogConsumer(LOG);
397        final Slf4jLogConsumer stderrConsumer = new Slf4jLogConsumer(LOG);
398
399        try (var callback = new FrameConsumerResultCallback()) {
400            callback.addConsumer(OutputFrame.OutputType.STDOUT, stdoutConsumer);
401            callback.addConsumer(OutputFrame.OutputType.STDERR, stderrConsumer);
402
403            getDockerClient().logContainerCmd(containerId)
404                .withStdOut(true)
405                .withStdErr(true)
406                .exec(callback)
407                .awaitCompletion();
408        } catch (InterruptedException e) {
409            Thread.currentThread().interrupt();
410        } catch (IOException e) {
411            LOG.warn("Could not read the output of container {}: {}", containerId, e.getMessage());
412        }
413    }
414
415    /**
416     * Copies a regular file out of a container, which may have exited.
417     *
418     * @param containerId the container
419     * @param containerPath the absolute path of the file in the container
420     * @param target the file to write
421     * @throws IOException if the file cannot be copied
422     * @since 5.1.0
423     */
424    @Internal
425    public void copyFileFromContainer(String containerId, String containerPath, Path target) throws IOException {
426        try (InputStream archive = getDockerClient().copyArchiveFromContainerCmd(containerId, containerPath).exec();
427             var tar = new TarArchiveInputStream(archive)) {
428            TarArchiveEntry entry = tar.getNextEntry();
429            if (entry == null || !entry.isFile()) {
430                throw new IOException(containerPath + " is not a file in container " + containerId);
431            }
432            Files.createDirectories(target.toAbsolutePath().getParent());
433            Files.copy(tar, target, StandardCopyOption.REPLACE_EXISTING);
434        } catch (NotFoundException e) {
435            throw new IOException(containerPath + " does not exist in container " + containerId, e);
436        }
437    }
438
439    /**
440     * Removes a container, stopping it if it is running.
441     *
442     * @param containerId the container
443     * @since 5.1.0
444     */
445    @Internal
446    public void removeContainer(String containerId) {
447        try (RemoveContainerCmd remove = getDockerClient().removeContainerCmd(containerId)) {
448            remove.withForce(true).withRemoveVolumes(true).exec();
449        } catch (NotFoundException _) {
450            LOG.debug("Container {} was already removed", containerId);
451        }
452    }
453
454    /**
455     * Removes an image reference without force: a tag, or the ID of an image that at most one tag refers to. Docker
456     * deletes the image when no tag refers to it anymore, and fails instead of removing an image that other tags refer
457     * to or that a container uses.
458     *
459     * @param image the tag or the image ID
460     * @since 5.1.0
461     */
462    @Internal
463    public void removeImage(String image) {
464        try (RemoveImageCmd remove = getDockerClient().removeImageCmd(image)) {
465            remove.exec();
466        } catch (NotFoundException _) {
467            LOG.debug("Image {} was already removed", image);
468        }
469    }
470
471    /**
472     * @param image the image name or ID
473     * @return the image details
474     * @since 5.1.0
475     */
476    @Internal
477    public InspectImageResponse inspectImage(String image) {
478        verifyDockerRunning();
479        try (InspectImageCmd inspect = getDockerClient().inspectImageCmd(image)) {
480            return inspect.exec();
481        }
482    }
483
484    /**
485     * @return the platform of the Docker daemon, as {@code os/architecture} with Go architecture names such as
486     * {@code linux/amd64} or {@code linux/arm64}
487     * @throws IllegalStateException if the Docker daemon is not reachable
488     * @since 5.1.0
489     */
490    @Internal
491    public String getDaemonPlatform() {
492        verifyDockerRunning();
493        Info info = getDockerClient().infoCmd().exec();
494        return info.getOsType() + "/" + goArchitecture(info.getArchitecture());
495    }
496
497    /**
498     * @param architecture an architecture name, as the Docker daemon or Jib configuration gives it
499     * @return the Go architecture name used in image platforms, such as {@code amd64} or {@code arm64}
500     */
501    static String goArchitecture(String architecture) {
502        return switch (architecture) {
503            case "x86_64", "amd64" -> "amd64";
504            case "aarch64", "arm64" -> "arm64";
505            default -> architecture;
506        };
507    }
508
509    /**
510     * Copies a file from the specified container path in the given image ID, into a temporal location.
511     *
512     * @param imageId The image ID.
513     * @param containerPath The container path.
514     * @return The temporal file.
515     */
516    public File copyFromContainer(String imageId, String containerPath) {
517        CreateContainerCmd containerCmd = getDockerClient().createContainerCmd(imageId);
518        CreateContainerResponse container = containerCmd.exec();
519        getDockerClient().startContainerCmd(container.getId());
520        InputStream nativeImage = getDockerClient().copyArchiveFromContainerCmd(container.getId(), containerPath).exec();
521
522        try (var fin = new TarArchiveInputStream(nativeImage)) {
523            TarArchiveEntry tarEntry = fin.getNextEntry();
524            File file = new File(mavenProject.getBuild().getDirectory(), tarEntry.getName());
525            if (!file.getCanonicalFile().toPath().startsWith(mavenProject.getBuild().getDirectory())) {
526                throw new IOException("Entry is outside of the target directory");
527            }
528
529            IOUtils.copy(fin, Files.newOutputStream(file.toPath()));
530
531            return file;
532        } catch (IOException e) {
533            LOG.error("Failed to copy file from container", e);
534        } finally {
535            containerCmd.close();
536        }
537        return null;
538    }
539
540    /**
541     * Loads the given Dockerfile as classpath resource and copies it into a temporary location in the target directory.
542     *
543     * @param name the name of the Dockerfile.
544     * @return the file where the Dockerfile was copied to.
545     */
546    public File loadDockerfileAsResource(String name) throws IOException {
547        return loadDockerfileAsResource(name, DockerfileMojo.DOCKERFILE);
548    }
549
550    /**
551     * Loads the given Dockerfile as classpath resource and copies it into a temporary location in the target directory.
552     *
553     * @param name the name of the Dockerfile.
554     * @param targetFileName the name of the file to copy the Dockerfile to.
555     * @return the file where the Dockerfile was copied to.
556     */
557    public File loadDockerfileAsResource(String name, String targetFileName) throws IOException {
558        String path = "/dockerfiles/" + name;
559        InputStream stream = getClass().getResourceAsStream(path);
560        if (stream != null) {
561            var dockerfile = new File(mavenProject.getBuild().getDirectory(), targetFileName);
562            FileUtils.copyInputStreamToFile(stream, dockerfile);
563            return dockerfile;
564        }
565        return null;
566    }
567
568    /**
569     * @param imageName the image name
570     * @return a {@link PushImageCmd} from the given image name.
571     */
572    public PushImageCmd pushImageCmd(String imageName) {
573        verifyDockerRunning();
574        return getDockerClient().pushImageCmd(imageName);
575    }
576
577    /**
578     * @param dockerImage the image name
579     * @param username the username
580     * @param password the password
581     * @return an {@link AuthConfig} object for the given image, username and password.
582     */
583    public AuthConfig getAuthConfigFor(String dockerImage, String username, String password) {
584        DockerImageName dockerImageName = DockerImageName.parse(dockerImage);
585        var defaultAuthConfig = new AuthConfig()
586            .withRegistryAddress(dockerImageName.getRegistry())
587            .withUsername(username)
588            .withPassword(password);
589        RegistryAuthLocator registryAuthLocator = RegistryAuthLocator.instance();
590        AuthConfig authConfig = registryAuthLocator.lookupAuthConfig(dockerImageName, defaultAuthConfig);
591        boolean loginSucceeded = false;
592        try {
593            AuthResponse authResponse = getDockerClient().authCmd().withAuthConfig(authConfig).exec();
594            if (authResponse.getStatus() != null && authResponse.getStatus().equals("Login Succeeded")) {
595                loginSucceeded = true;
596            }
597        } catch (Exception ignored) {
598            // typically this is com.github.dockerjava.api.exception.UnauthorizedException
599        }
600
601        if (loginSucceeded) {
602            LOG.info("Successfully logged in to registry {}", dockerImageName.getRegistry());
603        } else {
604            LOG.warn("Failed to login to registry {}", dockerImageName.getRegistry());
605        }
606        return authConfig;
607    }
608
609    private void verifyDockerRunning() {
610        try {
611            getDockerClient().pingCmd().exec();
612        } catch (DockerException e) {
613            throw new IllegalStateException(e.getMessage());
614        } catch (RuntimeException e) {
615            throw new IllegalStateException("Cannot connect to the Docker daemon at " + config.getDockerHost() + ". Is the docker daemon running?", e);
616        }
617    }
618
619    /**
620     * The exit code and the output of a container.
621     *
622     * @param exitCode the exit code
623     * @param output the standard output and error, interleaved
624     * @since 5.1.0
625     */
626    @Internal
627    public record ContainerOutput(int exitCode, String output) {
628    }
629
630    /**
631     * Splits the frames of a container stream into lines.
632     */
633    private static final class LineCallback extends ResultCallback.Adapter<Frame> {
634        private final Consumer<String> lines;
635        private final StringBuilder pending = new StringBuilder();
636
637        private LineCallback(Consumer<String> lines) {
638            this.lines = lines;
639        }
640
641        @Override
642        public void onNext(Frame frame) {
643            pending.append(new String(frame.getPayload(), StandardCharsets.UTF_8));
644            int newLine = pending.indexOf("\n");
645            while (newLine >= 0) {
646                lines.accept(StringUtils.removeEnd(pending.substring(0, newLine), "\r"));
647                pending.delete(0, newLine + 1);
648                newLine = pending.indexOf("\n");
649            }
650        }
651
652        private void flush() {
653            if (!pending.isEmpty()) {
654                lines.accept(pending.toString());
655                pending.setLength(0);
656            }
657        }
658    }
659}