@Test public void testSocketPeerConnectorNoCodec() throws Exception { final String portIdentifier = "abcd"; final ServerSocket serverSocket = new ServerSocket(0); final SiteToSiteClientConfig siteToSiteClientConfig = new SiteToSiteClientConfig(); siteToSiteClientConfig.setPortIdentifier(portIdentifier); final Peer peer = new Peer("localhost", 0, serverSocket.getLocalPort(), false, 0); final Future<Exception> future = startServer(serverSocket, peer, siteToSiteClientConfig, false); SocketPeerConnector connector = new SocketPeerConnector(peer, siteToSiteClientConfig, new SiteToSiteRemoteCluster()); SocketPeerConnection socketPeerConnection = connector.openConnection(false); Exception exception = future.get(); if (exception != null) { throw exception; } assertEquals(6, socketPeerConnection.getFlowFileProtocolVersion()); assertNull(socketPeerConnection.getFlowFileCodecVersion()); }
public SocketPeerConnection openConnection(boolean negotiateCodec) throws IOException { Socket socket; if (peer.isSecure()) { SSLContext sslContext = siteToSiteRemoteCluster.getSslContext(); if (sslContext == null) { throw new IOException("SSL not configured but peer is set to secure"); } socket = sslContext.getSocketFactory().createSocket(peer.getHostname(), peer.getRawPort()); } else { socket = new Socket(peer.getHostname(), peer.getRawPort()); } try { socket.setSoTimeout((int) siteToSiteClientConfig.getTimeout(TimeUnit.MILLISECONDS)); socket.getOutputStream().write(MAGIC_BYTES); int protocolVersion = negotiateVersion(socket, SOCKET_FLOW_FILE_PROTOCOL, new int[] {6, 5, 4, 3, 2, 1}); if (protocolVersion < 5 && siteToSiteClientConfig.getPortIdentifier() == null) { throw new IOException("Unable to find port identifier and it is required for this protocol version (" + protocolVersion + ")"); } protocolHandshake(socket, protocolVersion); Integer codecVersion = null; if (negotiateCodec) { RequestType.NEGOTIATE_FLOWFILE_CODEC.writeRequestType(new DataOutputStream(socket.getOutputStream())); codecVersion = negotiateVersion(socket, STANDARD_FLOW_FILE_CODEC, new int[] {1}); } return new SocketPeerConnection(socket, protocolVersion, codecVersion); } catch (IOException e) { socket.close(); throw e; } }
SocketPeerConnector { public SocketPeerConnection openConnection(boolean negotiateCodec) throws IOException { Socket socket; if (peer.isSecure()) { SSLContext sslContext = siteToSiteRemoteCluster.getSslContext(); if (sslContext == null) { throw new IOException("SSL not configured but peer is set to secure"); } socket = sslContext.getSocketFactory().createSocket(peer.getHostname(), peer.getRawPort()); } else { socket = new Socket(peer.getHostname(), peer.getRawPort()); } try { socket.setSoTimeout((int) siteToSiteClientConfig.getTimeout(TimeUnit.MILLISECONDS)); socket.getOutputStream().write(MAGIC_BYTES); int protocolVersion = negotiateVersion(socket, SOCKET_FLOW_FILE_PROTOCOL, new int[] {6, 5, 4, 3, 2, 1}); if (protocolVersion < 5 && siteToSiteClientConfig.getPortIdentifier() == null) { throw new IOException("Unable to find port identifier and it is required for this protocol version (" + protocolVersion + ")"); } protocolHandshake(socket, protocolVersion); Integer codecVersion = null; if (negotiateCodec) { RequestType.NEGOTIATE_FLOWFILE_CODEC.writeRequestType(new DataOutputStream(socket.getOutputStream())); codecVersion = negotiateVersion(socket, STANDARD_FLOW_FILE_CODEC, new int[] {1}); } return new SocketPeerConnection(socket, protocolVersion, codecVersion); } catch (IOException e) { socket.close(); throw e; } } }
SocketPeerConnector { public SocketPeerConnection openConnection(boolean negotiateCodec) throws IOException { Socket socket; if (peer.isSecure()) { SSLContext sslContext = siteToSiteRemoteCluster.getSslContext(); if (sslContext == null) { throw new IOException("SSL not configured but peer is set to secure"); } socket = sslContext.getSocketFactory().createSocket(peer.getHostname(), peer.getRawPort()); } else { socket = new Socket(peer.getHostname(), peer.getRawPort()); } try { socket.setSoTimeout((int) siteToSiteClientConfig.getTimeout(TimeUnit.MILLISECONDS)); socket.getOutputStream().write(MAGIC_BYTES); int protocolVersion = negotiateVersion(socket, SOCKET_FLOW_FILE_PROTOCOL, new int[] {6, 5, 4, 3, 2, 1}); if (protocolVersion < 5 && siteToSiteClientConfig.getPortIdentifier() == null) { throw new IOException("Unable to find port identifier and it is required for this protocol version (" + protocolVersion + ")"); } protocolHandshake(socket, protocolVersion); Integer codecVersion = null; if (negotiateCodec) { RequestType.NEGOTIATE_FLOWFILE_CODEC.writeRequestType(new DataOutputStream(socket.getOutputStream())); codecVersion = negotiateVersion(socket, STANDARD_FLOW_FILE_CODEC, new int[] {1}); } return new SocketPeerConnection(socket, protocolVersion, codecVersion); } catch (IOException e) { socket.close(); throw e; } } SocketPeerConnector(Peer peer, SiteToSiteClientConfig siteToSiteClientConfig, SiteToSiteRemoteCluster siteToSiteRemoteCluster); }
SocketPeerConnector { public SocketPeerConnection openConnection(boolean negotiateCodec) throws IOException { Socket socket; if (peer.isSecure()) { SSLContext sslContext = siteToSiteRemoteCluster.getSslContext(); if (sslContext == null) { throw new IOException("SSL not configured but peer is set to secure"); } socket = sslContext.getSocketFactory().createSocket(peer.getHostname(), peer.getRawPort()); } else { socket = new Socket(peer.getHostname(), peer.getRawPort()); } try { socket.setSoTimeout((int) siteToSiteClientConfig.getTimeout(TimeUnit.MILLISECONDS)); socket.getOutputStream().write(MAGIC_BYTES); int protocolVersion = negotiateVersion(socket, SOCKET_FLOW_FILE_PROTOCOL, new int[] {6, 5, 4, 3, 2, 1}); if (protocolVersion < 5 && siteToSiteClientConfig.getPortIdentifier() == null) { throw new IOException("Unable to find port identifier and it is required for this protocol version (" + protocolVersion + ")"); } protocolHandshake(socket, protocolVersion); Integer codecVersion = null; if (negotiateCodec) { RequestType.NEGOTIATE_FLOWFILE_CODEC.writeRequestType(new DataOutputStream(socket.getOutputStream())); codecVersion = negotiateVersion(socket, STANDARD_FLOW_FILE_CODEC, new int[] {1}); } return new SocketPeerConnection(socket, protocolVersion, codecVersion); } catch (IOException e) { socket.close(); throw e; } } SocketPeerConnector(Peer peer, SiteToSiteClientConfig siteToSiteClientConfig, SiteToSiteRemoteCluster siteToSiteRemoteCluster); SocketPeerConnection openConnection(boolean negotiateCodec); }
SocketPeerConnector { public SocketPeerConnection openConnection(boolean negotiateCodec) throws IOException { Socket socket; if (peer.isSecure()) { SSLContext sslContext = siteToSiteRemoteCluster.getSslContext(); if (sslContext == null) { throw new IOException("SSL not configured but peer is set to secure"); } socket = sslContext.getSocketFactory().createSocket(peer.getHostname(), peer.getRawPort()); } else { socket = new Socket(peer.getHostname(), peer.getRawPort()); } try { socket.setSoTimeout((int) siteToSiteClientConfig.getTimeout(TimeUnit.MILLISECONDS)); socket.getOutputStream().write(MAGIC_BYTES); int protocolVersion = negotiateVersion(socket, SOCKET_FLOW_FILE_PROTOCOL, new int[] {6, 5, 4, 3, 2, 1}); if (protocolVersion < 5 && siteToSiteClientConfig.getPortIdentifier() == null) { throw new IOException("Unable to find port identifier and it is required for this protocol version (" + protocolVersion + ")"); } protocolHandshake(socket, protocolVersion); Integer codecVersion = null; if (negotiateCodec) { RequestType.NEGOTIATE_FLOWFILE_CODEC.writeRequestType(new DataOutputStream(socket.getOutputStream())); codecVersion = negotiateVersion(socket, STANDARD_FLOW_FILE_CODEC, new int[] {1}); } return new SocketPeerConnection(socket, protocolVersion, codecVersion); } catch (IOException e) { socket.close(); throw e; } } SocketPeerConnector(Peer peer, SiteToSiteClientConfig siteToSiteClientConfig, SiteToSiteRemoteCluster siteToSiteRemoteCluster); SocketPeerConnection openConnection(boolean negotiateCodec); static final byte[] MAGIC_BYTES; static final int RESOURCE_OK; static final int DIFFERENT_RESOURCE_VERSION; static final int ABORT; static final String SOCKET_FLOW_FILE_PROTOCOL; static final String GZIP; static final String PORT_IDENTIFIER; static final String REQUEST_EXPIRATION_MILLIS; static final String BATCH_COUNT; static final String BATCH_SIZE; static final String BATCH_DURATION; static final String STANDARD_FLOW_FILE_CODEC; }