content: /*
   * Licensed to the Apache Software Foundation (ASF) under one
   * or more contributor license agreements.  See the NOTICE file
   * distributed with this work for additional information
   * regarding copyright ownership.  The ASF licenses this file
   * to you under the Apache License, Version 2.0 (the
   * "License"); you may not use this file except in compliance
   * with the License.  You may obtain a copy of the License at
   *
   *     https://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  package org.apache.avro;
  
  import static org.junit.jupiter.api.Assertions.*;
  
  import java.io.File;
  import java.io.FileReader;
  import java.io.FileWriter;
  import java.io.IOException;
  import java.io.LineNumberReader;
  import java.net.InetSocketAddress;
  import java.net.URL;
  import java.nio.ByteBuffer;
  import java.nio.file.Files;
  import java.util.ArrayList;
  import java.util.HashSet;
  import java.util.LinkedHashMap;
  import java.util.List;
  import java.util.Objects;
  import java.util.Random;
  import org.apache.avro.generic.GenericData;
  import org.apache.avro.generic.GenericRecord;
  import org.apache.avro.ipc.HttpTransceiver;
  import org.apache.avro.ipc.RPCContext;
  import org.apache.avro.ipc.RPCPlugin;
  import org.apache.avro.ipc.Requestor;
  import org.apache.avro.ipc.Responder;
  import org.apache.avro.ipc.Server;
  import org.apache.avro.ipc.SocketServer;
  import org.apache.avro.ipc.SocketTransceiver;
  import org.apache.avro.ipc.Transceiver;
  import org.apache.avro.ipc.generic.GenericRequestor;
  import org.apache.avro.ipc.specific.SpecificRequestor;
  import org.apache.avro.ipc.specific.SpecificResponder;
  import org.apache.avro.specific.SpecificData;
  import org.apache.avro.test.Kind;
  import org.apache.avro.test.MD5;
  import org.apache.avro.test.Simple;
  import org.apache.avro.test.TestError;
  import org.apache.avro.test.TestRecord;
  import org.junit.jupiter.api.AfterAll;
  import org.junit.jupiter.api.BeforeEach;
  import org.junit.jupiter.api.Test;
  
  public class TestProtocolSpecific {
  
    protected static final int REPEATING = -1;
  
    public static int ackCount;
  
    private static boolean throwUndeclaredError;
  
    public static class TestImpl implements Simple {
      public String hello(String greeting) {
        return "goodbye";
      }
  
      public int add(int arg1, int arg2) {
        return arg1 + arg2;
      }
  
      public TestRecord echo(TestRecord record) {
        return record;
      }
  
      public ByteBuffer echoBytes(ByteBuffer data) {
        return data;
      }
  
      public void error() throws TestError {
        if (throwUndeclaredError)
          throw new RuntimeException("foo");
        throw TestError.newBuilder().setMessage$("an error").build();
      }
  
      public void ack() {
        ackCount++;
      }
    }
  
    protected static Server server;
    protected static Transceiver client;
    protected static Simple proxy;
  
    protected static SpecificResponder responder;
  
    protected static HandshakeMonitor monitor;
  
    @BeforeEach
    public void testStartServer() throws Exception {
      if (server != null)
        return;
      responder = new SpecificResponder(Simple.class, new TestImpl());
      server = createServer(responder);
      server.start();
  
      client = createTransceiver();
      SpecificRequestor req = new SpecificRequestor(Simple.class, client);
      addRpcPlugins(req);
      proxy = SpecificRequestor.getClient(Simple.class, req);
  
      monitor = new HandshakeMonitor();
      responder.addRPCPlugin(monitor);
    }
  
    public void addRpcPlugins(Requestor requestor) {
    }
  
    public Server createServer(Responder testResponder) throws Exception {
      return server = new SocketServer(testResponder, new InetSocketAddress(0));
    }
  
    public Transceiver createTransceiver() throws Exception {
      return new SocketTransceiver(new InetSocketAddress(server.getPort()));
    }
  
    @Test
    void classLoader() throws Exception {
      ClassLoader loader = new ClassLoader() {
      };
  
      SpecificResponder responder = new SpecificResponder(Simple.class, new TestImpl(), new SpecificData(loader));
      assertEquals(responder.getSpecificData().getClassLoader(), loader);
  
      SpecificRequestor requestor = new SpecificRequestor(Simple.class, client, new SpecificData(loader));
      assertEquals(requestor.getSpecificData().getClassLoader(), loader);
    }
  
    @Test
    void getRemote() throws IOException {
      assertEquals(Simple.PROTOCOL, SpecificRequestor.getRemote(proxy));
    }
  
    @Test
    void hello() throws IOException {
      String response = proxy.hello("bob");
      assertEquals("goodbye", response);
    }
  
    @Test
    void testHashCode() throws IOException {
      TestError error = new TestError();
      error.hashCode();
    }
  
    @Test
    void echo() throws IOException {
      TestRecord record = new TestRecord();
      record.setName("foo");
      record.setKind(Kind.BAR);
      record.setHash(new MD5(new byte[] { 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5 }));
      TestRecord echoed = proxy.echo(record);
      assertEquals(record, echoed);
      assertEquals(record.hashCode(), echoed.hashCode());
    }
  
    @Test
    void add() throws IOException {
      int result = proxy.add(1, 2);
      assertEquals(3, result);
    }
  
    @Test
    void echoBytes() throws IOException {
      Random random = new Random();
      int length = random.nextInt(1024 * 16);
      ByteBuffer data = ByteBuffer.allocate(length);
      random.nextBytes(data.array());
      data.flip();
      ByteBuffer echoed = proxy.echoBytes(data);
      assertEquals(data, echoed);
    }
  
    @Test
    void emptyEchoBytes() throws IOException {
      ByteBuffer data = ByteBuffer.allocate(0);
      ByteBuffer echoed = proxy.echoBytes(data);
      data.flip();
      assertEquals(data, echoed);
    }
  
    @Test
    void error() throws IOException {
      TestError error = null;
      try {
        proxy.error();
      } catch (TestError e) {
        error = e;
      }
      assertNotNull(error);
      assertEquals("an error", error.getMessage$());
    }
  
    @Test
    void undeclaredError() throws Exception {
      this.throwUndeclaredError = true;
      RuntimeException error = null;
      try {
        proxy.error();
      } catch (RuntimeException e) {
        error = e;
      } finally {
        this.throwUndeclaredError = false;
      }
      assertNotNull(error);
      assertTrue(error.toString().contains("foo"));
    }
  
    @Test
    void oneWay() throws IOException {
      ackCount = 0;
      proxy.ack();
      proxy.hello("foo"); // intermix normal req
      proxy.ack();
      try {
        Thread.sleep(100);
      } catch (InterruptedException e) {
      }
      assertEquals(2, ackCount);
    }
  
    @Test
    void repeatedAccess() throws Exception {
      for (int x = 0; x < 1000; x++) {
        proxy.hello("hi!");
      }
    }
  
    @Test
    void connectionRefusedOneWay() throws IOException {
      assertThrows(Exception.class, () -> {
        Transceiver client = new HttpTransceiver(new URL("http://localhost:4444"));
        SpecificRequestor req = new SpecificRequestor(Simple.class, client);
        addRpcPlugins(req);
        Simple proxy = SpecificRequestor.getClient(Simple.class, req);
        proxy.ack();
      });
    }
  
    /**
     * Construct and use a protocol whose "hello" method has an extra argument to
     * check that schema is sent to parse request.
     */
    @Test
    void paramVariation() throws Exception {
      Protocol protocol = new Protocol("Simple", "org.apache.avro.test");
      List<Schema.Field> fields = new ArrayList<>();
      fields.add(new Schema.Field("extra", Schema.create(Schema.Type.BOOLEAN), null, null));
      fields.add(new Schema.Field("greeting", Schema.create(Schema.Type.STRING), null, null));
      Protocol.Message message = protocol.createMessage("hello", null /* doc */, new LinkedHashMap<String, String>(),
          Schema.createRecord(fields), Schema.create(Schema.Type.STRING), Schema.createUnion(new ArrayList<>()));
      protocol.getMessages().put("hello", message);
      try (Transceiver t = createTransceiver()) {
        GenericRequestor r = new GenericRequestor(protocol, t);
        addRpcPlugins(r);
        GenericRecord params = new GenericData.Record(message.getRequest());
        params.put("extra", Boolean.TRUE);
        params.put("greeting", "bob");
        String response = r.request("hello", params).toString();
        assertEquals("goodbye", response);
      }
    }
  
    @AfterAll
    public static void testHandshakeCount() throws IOException {
      monitor.assertHandshake();
    }
  
    @AfterAll
    public static void testStopServer() throws IOException {
      client.close();
      server.close();
      server = null;
    }
  
    public class HandshakeMonitor extends RPCPlugin {
  
      private int handshakes;
      private HashSet<String> seenProtocols = new HashSet<>();
  
      @Override
      public void serverConnecting(RPCContext context) {
        handshakes++;
        int expected = getExpectedHandshakeCount();
        if (expected > 0 && handshakes > expected) {
          throw new IllegalStateException(
              "Expected number of Protocol negotiation handshakes exceeded expected " + expected + " was " + handshakes);
        }
  
        // check that a given client protocol is only sent once
        String clientProtocol = context.getHandshakeRequest().getClientProtocol();
        if (clientProtocol != null) {
          assertFalse(seenProtocols.contains(clientProtocol));
          seenProtocols.add(clientProtocol);
        }
      }
  
      public void assertHandshake() {
        int expected = getExpectedHandshakeCount();
        if (expected != REPEATING) {
          assertEquals(expected, handshakes, "Expected number of handshakes did not take place.");
        }
      }
    }
  
    protected int getExpectedHandshakeCount() {
      return 3;
    }
  
    public static class InteropTest {
  
      private static File SERVER_PORTS_DIR;
      static {
        try {
          SERVER_PORTS_DIR = Files.createTempDirectory(TestProtocolSpecific.class.getSimpleName()).toFile();
        } catch (IOException e) {
          e.printStackTrace();
        }
      }
  
      @Test
      void client() throws Exception {
        for (File f : Objects.requireNonNull(SERVER_PORTS_DIR.listFiles())) {
          try (LineNumberReader reader = new LineNumberReader(new FileReader(f))) {
            int port = Integer.parseInt(reader.readLine());
            System.out.println("Validating java client to " + f.getName() + " - " + port);
            Transceiver client = new SocketTransceiver(new InetSocketAddress("localhost", port));
            proxy = SpecificRequestor.getClient(Simple.class, client);
            TestProtocolSpecific proto = new TestProtocolSpecific();
            proto.hello();
            proto.echo();
            proto.echoBytes();
            proto.error();
            System.out.println("Done! Validation java client to " + f.getName() + " - " + port);
          }
        }
      }
  
      /**
       * Starts the RPC server.
       */
      public static void main(String[] args) throws Exception {
        SocketServer server = new SocketServer(new SpecificResponder(Simple.class, new TestImpl()),
            new InetSocketAddress(0));
        server.start();
        File portFile = new File(SERVER_PORTS_DIR, "java-port");
        try (FileWriter w = new FileWriter(portFile)) {
          w.write(Integer.toString(server.getPort()));
        }
  
      }
    }
  }
  