@@ -0,0 +1,3125 @@
+/*%0A * Licensed to the Apache Software Foundation (ASF) under one or more%0A * contributor license agreements.  See the NOTICE file distributed with%0A * this work for additional information regarding copyright ownership.%0A * The ASF licenses this file to You under the Apache License, Version 2.0%0A * (the %22License%22); you may not use this file except in compliance with%0A * the License.  You may obtain a copy of the License at%0A *%0A *      http://www.apache.org/licenses/LICENSE-2.0%0A *%0A * Unless required by applicable law or agreed to in writing, software%0A * distributed under the License is distributed on an %22AS IS%22 BASIS,%0A * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.%0A * See the License for the specific language governing permissions and%0A * limitations under the License.%0A */%0Apackage org.apache.beam.runners.extensions.metrics;%0A%0Aimport java.io.BufferedReader;%0Aimport java.io.IOException;%0Aimport java.io.InputStream;%0Aimport java.io.InputStreamReader;%0Aimport java.net.ServerSocket;%0Aimport java.net.Socket;%0Aimport java.nio.charset.Charset;%0Aimport java.util.Collection;%0Aimport java.util.concurrent.CopyOnWriteArrayList;%0Aimport java.util.concurrent.atomic.AtomicBoolean;%0A%0A/** Mock of a network server. */%0Aclass NetworkMockServer %7B%0A  private final int port;%0A%0A  private ServerSocket server;%0A  private GraphiteThread thread;%0A%0A  private Collection%3CString%3E messages = new CopyOnWriteArrayList%3CString%3E();%0A%0A  public NetworkMockServer(final int port) %7B%0A    this.port = port;%0A  %7D%0A%0A  public int getPort() %7B%0A    return port;%0A  %7D%0A%0A  public NetworkMockServer start() throws IOException %7B%0A    server = new ServerSocket(port);%0A    thread = new GraphiteThread(server, messages);%0A    thread.start();%0A    return this;%0A  %7D%0A%0A  public void stop() throws IOException %7B%0A    thread.shutdown();%0A    server.close();%0A  %7D%0A%0A  public Collection%3CString%3E getMessages() %7B%0A    return messages;%0A  %7D%0A%0A  public void clear() %7B%0A    messages.clear();%0A  %7D%0A%0A  private static class GraphiteThread extends Thread %7B%0A    private final Collection%3CString%3E messages;%0A%0A    private final AtomicBoolean done = new AtomicBoolean(false);%0A    private final ServerSocket server;%0A%0A    public GraphiteThread(final ServerSocket server, final Collection%3CString%3E messages) %7B%0A      this.messages = messages;%0A      this.server = server;%0A      setName(%22network-mock-server%22);%0A    %7D%0A%0A    @Override%0A    public void run() %7B%0A      while (!done.get()) %7B%0A        try %7B%0A          final Socket s = server.accept();%0A          synchronized (this) %7B%0A            final InputStream is = s.getInputStream();%0A            final BufferedReader reader =%0A                new BufferedReader(new InputStreamReader(is, Charset.forName(%22UTF-8%22)));%0A            String line;%0A%0A            try %7B%0A              while ((line = reader.readLine()) != null) %7B%0A                messages.add(line);%0A              %7D%0A            %7D finally %7B%0A              s.close();%0A            %7D%0A          %7D%0A        %7D catch (final IOException e) %7B%0A          if (!done.get()) %7B%0A            throw new RuntimeException(e);%0A          %7D%0A        %7D%0A      %7D%0A    %7D%0A%0A    public void shutdown() %7B%0A      done.set(true);%0A    %7D%0A  %7D%0A%7D%0A

Introduce NetworkMockServer to be used for all metrics sinks tests