1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
|
/*
* Copyright (C) 2021 Samsung Electronics
* Licensed 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
* http://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.onap.a1pesimulator.service.fileready;
import static java.util.Objects.nonNull;
import static org.onap.a1pesimulator.util.Constants.TEMP_DIR;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.nio.file.Files;
import java.util.zip.GZIPOutputStream;
import org.onap.a1pesimulator.data.fileready.FileData;
import org.onap.a1pesimulator.exception.NotUploadedToFtpException;
import org.onap.a1pesimulator.util.VnfConfigReader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.util.FileCopyUtils;
import net.schmizz.sshj.SSHClient;
import net.schmizz.sshj.sftp.SFTPClient;
import net.schmizz.sshj.transport.verification.PromiscuousVerifier;
import reactor.core.publisher.Mono;
@Service
public class FtpServerService {
private static final Logger log = LoggerFactory.getLogger(FtpServerService.class);
//true - file will be uploaded to FTP; false - file will be copied into xmlPmLocation
@Value("${ftp.server.upload}")
private boolean ftpServerUpload;
// location where archived file will be copied
@Value("${xml.pm.location}")
private String xmlPmLocation;
@Value("${ftp.server.protocol}")
private String ftpServerProtocol;
@Value("${ftp.server.filepath}")
private String ftpServerFilepath;
@Value("${ftp.server.username}")
private String ftpServerUsername;
@Value("${ftp.server.password}")
private String ftpServerPassword;
private VnfConfigReader vnfConfigReader;
public FtpServerService(VnfConfigReader vnfConfigReader) {
this.vnfConfigReader = vnfConfigReader;
}
public Mono<FileData> uploadFileToFtp(FileData fileData) {
return Mono.just(fileData)
.flatMap(this::tryToCompressFile)
.flatMap(this::tryToUploadOrSaveFileToFtp)
.onErrorResume(throwable -> resumeError(throwable, fileData))
.doOnNext(file -> deletePMBulkFile(file.getPmBulkFile()));
}
/**
* Trying to compress file into .gz
*
* @param fileData file to be archived
* @return archived file
*/
private Mono<FileData> tryToCompressFile(FileData fileData) {
File archiveBulkFile = new File(TEMP_DIR, fileData.getPmBulkFile().getName() + ".gz");
try (GZIPOutputStream zos = new GZIPOutputStream(
new FileOutputStream(archiveBulkFile.getAbsolutePath())); FileInputStream inputStream = new FileInputStream(fileData.getPmBulkFile())) {
byte[] buffer = new byte[1024];
int len;
while ((len = inputStream.read(buffer)) > 0) {
zos.write(buffer, 0, len);
}
fileData.setArchivedPmBulkFile(archiveBulkFile);
log.trace("Compressing file {}", fileData.getPmBulkFile().getName());
return Mono.just(fileData);
} catch (IOException e) {
log.error("Could not compress file", e);
return Mono.empty();
}
}
/**
* Upload file to FTP or copy it to mounted location
*
* @param fileData data about file
* @return fileData for fileReadyEvent
*/
private Mono<FileData> tryToUploadOrSaveFileToFtp(FileData fileData) {
if (ftpServerUpload) {
return tryToUploadFileToFtp(fileData);
} else {
File fileOnFtp = new File(xmlPmLocation, fileData.getArchivedPmBulkFile().getName());
try {
FileCopyUtils.copy(fileData.getArchivedPmBulkFile(), fileOnFtp);
log.info("Uploading file to the location: {}", fileOnFtp);
return Mono.just(fileData);
} catch (IOException e) {
return Mono.error(new NotUploadedToFtpException("File was not copied to FTP location", e));
}
}
}
/**
* Upload file to FTP
*
* @param fileData archived file in Mono
* @return archived file for fileReadyEvent
*/
private Mono<FileData> tryToUploadFileToFtp(FileData fileData) {
if (nonNull(fileData.getArchivedPmBulkFile())) {
SSHClient client = getSSHClient();
if (nonNull(client)) {
try (client; SFTPClient sftpClient = client.newSFTPClient()) {
File archiveBulkFile = fileData.getArchivedPmBulkFile();
sftpClient.put(archiveBulkFile.getAbsolutePath(), ftpServerFilepath + "/" + archiveBulkFile.getName());
log.info("Uploading file to FTP: {}", archiveBulkFile.getAbsoluteFile());
return Mono.just(fileData);
} catch (IOException e) {
log.error("Exception while trying to upload a file", e);
}
} else {
log.error("Could not connect to FTP server");
}
} else {
log.error("There is no file to upload");
}
return Mono.error(new NotUploadedToFtpException("File was not uploaded to FTP"));
}
/**
* Creates SSHClient instance
*
* @return SSHClient
*/
protected SSHClient getSSHClient() {
SSHClient client = new SSHClient();
try {
client.addHostKeyVerifier(new PromiscuousVerifier());
client.connect(vnfConfigReader.getVnfConfig().getFtpHost(), Integer.parseInt(vnfConfigReader.getVnfConfig().getFtpPort()));
client.authPassword(ftpServerUsername, ftpServerPassword);
return client;
} catch (IOException e) {
log.error("There was an error while connecting to FTP server", e);
try {
client.close();
} catch (IOException ioException) {
log.error("There was an error while closing the connection to FTP server", e);
}
return null;
}
}
/**
* Deletes created PM Bulk File xml from temp storage after successful upload to FTP
*
* @param file file which we gonna delete
*/
public static void deletePMBulkFile(File file) {
try {
log.trace("Deleting file: {}", file.getAbsoluteFile());
Files.delete(file.toPath());
} catch (IOException e) {
log.warn("Could not delete file: {}", file.getName(), e);
}
}
/**
* Get path to FTP server
*
* @return for example: "sftp://foo:pass@106.120.119.170:2222/upload/"
*/
public String getFtpPath() {
return ftpServerProtocol + "://" + ftpServerUsername + ":" + ftpServerPassword + "@" + vnfConfigReader.getVnfConfig().getFtpHost() + ":"
+ vnfConfigReader.getVnfConfig().getFtpPort() + "/" + ftpServerFilepath
+ "/";
}
/**
* Try to clean up things after an exception
*
* @param throwable error thrown
* @param fileData data about files which needs to be deleted
* @return empty Mono object
*/
protected Mono<FileData> resumeError(Throwable throwable, FileData fileData) {
log.error("Error occurs while uploading file to FTP server", throwable);
deletePMBulkFile(fileData.getPmBulkFile());
deletePMBulkFile(fileData.getArchivedPmBulkFile());
return Mono.empty();
}
}
|
