Skip to content

Commit ca74218

Browse files
gem-neo4jgem-neo4j
committed
Reduce Common Usage between APOC Extended and APOC Core
1 parent 444989b commit ca74218

File tree

67 files changed

+316
-188
lines changed

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

67 files changed

+316
-188
lines changed

‎common/src/main/java/apoc/ApocConfig.java

Lines changed: 39 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,6 @@
1818
*/
1919
package apoc;
2020

21-
import static apoc.util.FileUtils.isFile;
2221
import static java.lang.String.format;
2322
import static org.neo4j.configuration.BootloaderSettings.lib_directory;
2423
import static org.neo4j.configuration.BootloaderSettings.run_directory;
@@ -32,9 +31,14 @@
3231
import static org.neo4j.internal.helpers.ProcessUtils.executeCommandWithOutput;
3332

3433
import apoc.export.util.ExportConfig;
34+
import apoc.util.FileUtils;
35+
import apoc.util.SupportedProtocols;
36+
import apoc.util.Util;
3537
import java.io.File;
3638
import java.io.IOException;
3739
import java.lang.reflect.Field;
40+
import java.net.MalformedURLException;
41+
import java.net.URI;
3842
import java.net.URL;
3943
import java.nio.file.Path;
4044
import java.time.Duration;
@@ -337,6 +341,40 @@ public void checkWriteAllowed(ExportConfig exportConfig, String fileName) {
337341
}
338342
}
339343

344+
public static boolean isFile(String fileName) {
345+
return from(fileName) == SupportedProtocols.file;
346+
}
347+
348+
public static SupportedProtocols from(URL url) {
349+
return FileUtils.of(url.getProtocol());
350+
}
351+
352+
public static SupportedProtocols from(String source) {
353+
try {
354+
final URL url = new URL(source);
355+
return from(url);
356+
} catch (MalformedURLException e) {
357+
if (!e.getMessage().contains("no protocol")) {
358+
try {
359+
// in case new URL(source) throw e.g. unknown protocol: hdfs, because of missing jar,
360+
// we retrieve the related enum and throw the associated MissingDependencyException(..)
361+
// otherwise we return unknown protocol: yyyyy
362+
return SupportedProtocols.valueOf(new URI(source).getScheme());
363+
} catch (Exception ignored) {
364+
}
365+
366+
// in case a Windows user write an url like `C:/User/...`
367+
if (e.getMessage().contains("unknown protocol") && Util.isWindows()) {
368+
throw new RuntimeException(e.getMessage()
369+
+ "\n Please note that for Windows absolute paths they have to be explicit by prepending `file:` or supplied without the drive, "
370+
+ "\n e.g. `file:C:/my/path/file` or `/my/path/file`, instead of `C:/my/path/file`");
371+
}
372+
throw new RuntimeException(e);
373+
}
374+
return SupportedProtocols.file;
375+
}
376+
}
377+
340378
public static ApocConfig apocConfig() {
341379
return theInstance;
342380
}

‎common/src/main/java/apoc/result/VirtualPath.java

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,7 @@
1818
*/
1919
package apoc.result;
2020

21-
import apoc.util.CollectionUtils;
21+
import apoc.util.Util;
2222
import java.util.ArrayList;
2323
import java.util.Arrays;
2424
import java.util.Iterator;
@@ -131,7 +131,7 @@ public String toString() {
131131

132132
private void requireConnected(Relationship relationship) {
133133
final List<Node> previousNodes = getPreviousNodes();
134-
boolean isRelConnectedToPrevious = CollectionUtils.containsAny(previousNodes, relationship.getNodes());
134+
boolean isRelConnectedToPrevious = Util.containsAny(previousNodes, relationship.getNodes());
135135
if (!isRelConnectedToPrevious) {
136136
throw new IllegalArgumentException("Relationship is not part of current path.");
137137
}

‎common/src/main/java/apoc/util/FileUtils.java

Lines changed: 151 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -20,9 +20,10 @@
2020

2121
import static apoc.ApocConfig.APOC_IMPORT_FILE_ALLOW__READ__FROM__FILESYSTEM;
2222
import static apoc.ApocConfig.apocConfig;
23+
import static apoc.export.util.LimitedSizeInputStream.toLimitedIStream;
2324
import static apoc.util.Util.ERROR_BYTES_OR_STRING;
2425
import static apoc.util.Util.REDIRECT_LIMIT;
25-
import static apoc.util.Util.readHttpInputStream;
26+
import static apoc.util.Util.isRedirect;
2627

2728
import apoc.ApocConfig;
2829
import apoc.export.util.CountingInputStream;
@@ -32,22 +33,32 @@
3233
import apoc.util.s3.S3URLConnection;
3334
import apoc.util.s3.S3UploadUtils;
3435
import java.io.BufferedOutputStream;
36+
import java.io.BufferedWriter;
37+
import java.io.ByteArrayInputStream;
3538
import java.io.File;
3639
import java.io.FileOutputStream;
3740
import java.io.IOException;
41+
import java.io.InputStream;
3842
import java.io.OutputStream;
43+
import java.io.OutputStreamWriter;
44+
import java.io.StringWriter;
45+
import java.net.HttpURLConnection;
3946
import java.net.MalformedURLException;
4047
import java.net.URI;
4148
import java.net.URISyntaxException;
4249
import java.net.URL;
50+
import java.net.URLConnection;
4351
import java.net.URLStreamHandler;
4452
import java.net.URLStreamHandlerFactory;
4553
import java.nio.file.NoSuchFileException;
4654
import java.nio.file.Path;
4755
import java.nio.file.Paths;
4856
import java.util.Map;
4957
import java.util.Optional;
58+
import org.apache.commons.compress.archivers.ArchiveEntry;
59+
import org.apache.commons.compress.archivers.ArchiveInputStream;
5060
import org.apache.commons.io.FilenameUtils;
61+
import org.apache.commons.io.IOUtils;
5162
import org.apache.commons.lang3.StringUtils;
5263
import org.neo4j.graphdb.security.URLAccessChecker;
5364
import org.neo4j.graphdb.security.URLAccessValidationError;
@@ -187,7 +198,7 @@ public static CountingInputStream inputStreamFor(
187198
if (input instanceof String) {
188199
String fileName = (String) input;
189200
fileName = changeFileUrlIfImportDirectoryConstrained(fileName, urlAccessChecker);
190-
return Util.openInputStream(fileName, headers, payload, compressionAlgo, urlAccessChecker);
201+
return FileUtils.openInputStream(fileName, headers, payload, compressionAlgo, urlAccessChecker);
191202
} else if (input instanceof byte[]) {
192203
return getInputStreamFromBinary((byte[]) input, compressionAlgo);
193204
} else {
@@ -345,4 +356,142 @@ public static File getLogDirectory() {
345356
public static CountingInputStream getInputStreamFromBinary(byte[] urlOrBinary, String compressionAlgo) {
346357
return CompressionAlgo.valueOf(compressionAlgo).toInputStream(urlOrBinary);
347358
}
359+
360+
public static StreamConnection readHttpInputStream(
361+
String urlAddress,
362+
Map<String, Object> headers,
363+
String payload,
364+
int redirectLimit,
365+
URLAccessChecker urlAccessChecker)
366+
throws IOException {
367+
URL url = ApocConfig.apocConfig().checkAllowedUrlAndPinToIP(urlAddress, urlAccessChecker);
368+
URLConnection con = openUrlConnection(url, headers);
369+
writePayload(con, payload);
370+
String newUrl = handleRedirect(con, urlAddress);
371+
if (newUrl != null && !urlAddress.equals(newUrl)) {
372+
con.getInputStream().close();
373+
if (redirectLimit == 0) {
374+
throw new IOException("Redirect limit exceeded");
375+
}
376+
return readHttpInputStream(newUrl, headers, payload, --redirectLimit, urlAccessChecker);
377+
}
378+
379+
return new StreamConnection.UrlStreamConnection(con);
380+
}
381+
382+
public static URLConnection openUrlConnection(URL src, Map<String, Object> headers) throws IOException {
383+
URLConnection con = src.openConnection();
384+
con.setRequestProperty("User-Agent", "APOC Procedures for Neo4j");
385+
if (con instanceof HttpURLConnection) {
386+
HttpURLConnection http = (HttpURLConnection) con;
387+
http.setInstanceFollowRedirects(false);
388+
if (headers != null) {
389+
Object method = headers.get("method");
390+
if (method != null) {
391+
http.setRequestMethod(method.toString());
392+
http.setChunkedStreamingMode(1024 * 1024);
393+
}
394+
headers.forEach((k, v) -> con.setRequestProperty(k, v == null ? "" : v.toString()));
395+
}
396+
}
397+
398+
con.setConnectTimeout(apocConfig().getInt("apoc.http.timeout.connect", 10_000));
399+
con.setReadTimeout(apocConfig().getInt("apoc.http.timeout.read", 60_000));
400+
return con;
401+
}
402+
403+
private static void writePayload(URLConnection con, String payload) throws IOException {
404+
if (payload == null) return;
405+
con.setDoOutput(true);
406+
BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(con.getOutputStream(), "UTF-8"));
407+
writer.write(payload);
408+
writer.close();
409+
}
410+
411+
private static String handleRedirect(URLConnection con, String url) throws IOException {
412+
if (!(con instanceof HttpURLConnection)) return url;
413+
if (!isRedirect(((HttpURLConnection) con))) return url;
414+
return con.getHeaderField("Location");
415+
}
416+
417+
public static CountingInputStream openInputStream(
418+
Object input,
419+
Map<String, Object> headers,
420+
String payload,
421+
String compressionAlgo,
422+
URLAccessChecker urlAccessChecker)
423+
throws IOException, URISyntaxException, URLAccessValidationError {
424+
if (input instanceof String) {
425+
String urlAddress = (String) input;
426+
final ArchiveType archiveType = ArchiveType.from(urlAddress);
427+
if (archiveType.isArchive()) {
428+
return getStreamCompressedFile(urlAddress, headers, payload, archiveType, urlAccessChecker);
429+
}
430+
431+
StreamConnection sc = getStreamConnection(urlAddress, headers, payload, urlAccessChecker);
432+
return sc.toCountingInputStream(compressionAlgo);
433+
} else if (input instanceof byte[]) {
434+
return FileUtils.getInputStreamFromBinary((byte[]) input, compressionAlgo);
435+
} else {
436+
throw new RuntimeException(ERROR_BYTES_OR_STRING);
437+
}
438+
}
439+
440+
private static CountingInputStream getStreamCompressedFile(
441+
String urlAddress,
442+
Map<String, Object> headers,
443+
String payload,
444+
ArchiveType archiveType,
445+
URLAccessChecker urlAccessChecker)
446+
throws IOException, URISyntaxException, URLAccessValidationError {
447+
StreamConnection sc;
448+
InputStream stream;
449+
String[] tokens = urlAddress.split("!");
450+
urlAddress = tokens[0];
451+
String zipFileName;
452+
if (tokens.length == 2) {
453+
zipFileName = tokens[1];
454+
sc = getStreamConnection(urlAddress, headers, payload, urlAccessChecker);
455+
stream = getFileStreamIntoCompressedFile(sc.getInputStream(), zipFileName, archiveType);
456+
stream = toLimitedIStream(stream, sc.getLength());
457+
} else throw new IllegalArgumentException("filename can't be null or empty");
458+
459+
return new CountingInputStream(stream, sc.getLength());
460+
}
461+
462+
public static StreamConnection getStreamConnection(
463+
String urlAddress, Map<String, Object> headers, String payload, URLAccessChecker urlAccessChecker)
464+
throws IOException, URISyntaxException, URLAccessValidationError {
465+
return FileUtils.getStreamConnection(
466+
FileUtils.from(urlAddress), urlAddress, headers, payload, urlAccessChecker);
467+
}
468+
469+
private static InputStream getFileStreamIntoCompressedFile(InputStream is, String fileName, ArchiveType archiveType)
470+
throws IOException {
471+
try (ArchiveInputStream archive = archiveType.getInputStream(is)) {
472+
ArchiveEntry archiveEntry;
473+
474+
while ((archiveEntry = archive.getNextEntry()) != null) {
475+
if (!archiveEntry.isDirectory() && archiveEntry.getName().equals(fileName)) {
476+
return new ByteArrayInputStream(IOUtils.toByteArray(archive));
477+
}
478+
}
479+
}
480+
481+
return null;
482+
}
483+
484+
public static Object getStringOrCompressedData(StringWriter writer, ExportConfig config) {
485+
try {
486+
final String compression = config.getCompressionAlgo();
487+
final String writerString = writer.toString();
488+
Object data = compression.equals(CompressionAlgo.NONE.name())
489+
? writerString
490+
: CompressionAlgo.valueOf(compression).compress(writerString, config.getCharset());
491+
writer.getBuffer().setLength(0);
492+
return data;
493+
} catch (Exception e) {
494+
throw new RuntimeException(e);
495+
}
496+
}
348497
}

0 commit comments

Comments
 (0)