This is an automated email from the ASF dual-hosted git repository. aiceflower pushed a commit to branch release-0.9.4 in repository https://gitbox.apache.org/repos/asf/linkis.git
commit cc83e11e6ee831d176f7352d5721bf2608f21fe1 Author: jftang <[email protected]> AuthorDate: Wed Jun 3 19:18:42 2020 +0800 rebuild pipe-engine module --- .../linkis/engine/pipeline/domain/PipeEntity.java | 148 --------------------- ...rrorException.scala => OutputStreamCache.scala} | 16 +-- .../{PipeTest.scala => PipeEngineJob.scala} | 23 ++-- .../engine/pipeline/PipeLineEngineExecutor.scala | 13 +- .../pipeline/PipeLineEngineExecutorFactory.scala | 6 +- .../{PipeTest.scala => PipelineEngineParser.scala} | 27 ++-- .../PipeLineConstant.scala} | 17 ++- .../FsConversions.scala} | 21 ++- .../exception/PipeLineErrorException.scala | 4 +- .../engine/pipeline/executor/CSVExecutor.scala | 69 +++++----- .../engine/pipeline/executor/CopyExecutor.scala | 43 +++--- .../engine/pipeline/executor/ExcelExecutor.scala | 73 +++++----- .../pipeline/executor/PipeLineExecutor.scala | 21 +-- .../executor/PipeLineExecutorFactory.scala | 8 +- .../engine/pipeline/executor/TxtExecutor.scala.bak | 48 ------- .../linkis/engine/pipeline/parser/IEParser.scala | 104 --------------- 16 files changed, 174 insertions(+), 467 deletions(-) diff --git a/ujes/definedEngines/pipeline/engine/src/main/java/com/webank/wedatasphere/linkis/engine/pipeline/domain/PipeEntity.java b/ujes/definedEngines/pipeline/engine/src/main/java/com/webank/wedatasphere/linkis/engine/pipeline/domain/PipeEntity.java deleted file mode 100644 index 0091fe422a..0000000000 --- a/ujes/definedEngines/pipeline/engine/src/main/java/com/webank/wedatasphere/linkis/engine/pipeline/domain/PipeEntity.java +++ /dev/null @@ -1,148 +0,0 @@ -/* - * Copyright 2019 WeBank - * - * 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 com.webank.wedatasphere.linkis.engine.pipeline.domain; - -import com.webank.wedatasphere.linkis.engine.pipeline.exception.PipeLineErrorException; -import com.webank.wedatasphere.linkis.engine.pipeline.parser.IEParser; -import org.apache.commons.lang.StringUtils; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.util.HashMap; -import java.util.Map; -import java.util.regex.Matcher; -import java.util.regex.Pattern; - -/** - * Created by johnnwang on 2018/11/14. - */ -public class PipeEntity { - - private final Logger LOGGER = LoggerFactory.getLogger(getClass()); - - public static final String SET_REGEX = "\\s*set\\s+([a-z]{2,12})\\s*=\\s*(\\S+)\\s?"; - public static final String STORAGE_REGEX = "(?i)\\s*from\\s+(\\S+)\\s+to\\s+(\\S+)\\s?"; - static final Pattern SET_PATTERN = Pattern.compile(SET_REGEX); - static final Pattern STORAGE_PATTERN = Pattern.compile(STORAGE_REGEX); - private String source; - private String dest; - private String type; - private Map<String, String> paramMap = new HashMap<String, String>(); - - public PipeEntity(String script) throws PipeLineErrorException { - - - /** - * script样式： - * %pipeLine.toWorkspace - * set key=value; - * set key=value; - * set key=value; - * set key=value; - * from source to dest; - */ - String kind = IEParser.getKind(script); - if ("pipeLine.toWorkspace".equals(kind)) { - this.type = "toWorkspace"; - } else if ("pipeLine.toHDFS".equals(kind)) { - this.type = "toHDFS"; - } else if ("pipeLine.toExcel".equals(kind)) { - this.type = "toExcel"; - } else if ("pipeLine.toCSV".equals(kind)) { - this.type = "toCSV"; - } else if ("pipeLine.toTxt".equals(kind)) { - this.type = "toTxt"; - } else { - // TODO: 2018/11/16 错误码待定 - LOGGER.error("not support(不支持) " + kind + " Import and export type(的导入导出类型)"); - throw new PipeLineErrorException(70001, "不支持" + kind + "的导入导出类型"); - } - String code = IEParser.getFormatCode(script); - if (StringUtils.isEmpty(code)) { - LOGGER.error("code is Empty!"); - throw new PipeLineErrorException(70002, "The execution code is empty!(执行代码为空!)"); - } - String splitRegex = ";"; - if (code.indexOf(";") < 0) { - splitRegex = "\n"; - } - String[] codes = code.split(splitRegex); - for (String _code : codes) { - Matcher matcher = SET_PATTERN.matcher(_code); - boolean find = matcher.find(); - if (find) { - String key = matcher.group(1); - String value = matcher.group(2); - paramMap.put(key, value); - //else ignore it. - continue; - } - matcher = STORAGE_PATTERN.matcher(_code); - find = matcher.find(); - if (find) { - source = matcher.group(1); - dest = matcher.group(2); - } //else ignore it. - } - if (StringUtils.isEmpty(source) || StringUtils.isEmpty(dest)) { - LOGGER.error("Import or export address is empty(导入或导出地址为空)"); - throw new PipeLineErrorException(70003, "Import or export address is empty（导入或导出地址为空）"); - } - } - - @Override - public String toString() { - return "PipeEntity{" + - "source='" + source + '\'' + - ", dest='" + dest + '\'' + - ", type='" + type + '\'' + - ", paramMap=" + paramMap + - '}'; - } - - public String getSource() { - return source; - } - - public void setSource(String source) { - this.source = source; - } - - public String getDest() { - return dest; - } - - public void setDest(String dest) { - this.dest = dest; - } - - public String getType() { - return type; - } - - public void setType(String type) { - this.type = type; - } - - public Map<String, String> getParamMap() { - return paramMap; - } - - public void setParamMap(Map<String, String> paramMap) { - this.paramMap = paramMap; - } -} diff --git a/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/exception/PipeLineErrorException.scala b/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/OutputStreamCache.scala similarity index 67% copy from ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/exception/PipeLineErrorException.scala copy to ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/OutputStreamCache.scala index 2f3a8ec394..70c4a7a633 100644 --- a/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/exception/PipeLineErrorException.scala +++ b/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/OutputStreamCache.scala @@ -13,14 +13,14 @@ * See the License for the specific language governing permissions and * limitations under the License. */ +package com.webank.wedatasphere.linkis.engine.pipeline -package com.webank.wedatasphere.linkis.engine.pipeline.exception - -import com.webank.wedatasphere.linkis.common.exception.ErrorException +import java.io.OutputStream /** - * Created by johnnwang on 2018/11/16. - */ -class PipeLineErrorException(errCode: Int, desc: String) extends ErrorException(errCode, desc) { - -} + * created by patinousward on 2019/12/9 + * Description: + */ +object OutputStreamCache { + final val osCache = new java.util.HashMap[String, OutputStream]() +} \ No newline at end of file diff --git a/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/PipeTest.scala b/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/PipeEngineJob.scala similarity index 55% copy from ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/PipeTest.scala copy to ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/PipeEngineJob.scala index a354a4a65b..ce9bdf1a08 100644 --- a/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/PipeTest.scala +++ b/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/PipeEngineJob.scala @@ -13,23 +13,18 @@ * See the License for the specific language governing permissions and * limitations under the License. */ - package com.webank.wedatasphere.linkis.engine.pipeline -import com.webank.wedatasphere.linkis.engine.pipeline.domain.PipeEntity -import com.webank.wedatasphere.linkis.engine.pipeline.util.PipeLineUtil +import com.webank.wedatasphere.linkis.engine.execute.CommonEngineJob /** - * Created by johnnwang on 2018/11/14. - */ -object PipeTest { - def main(args: Array[String]): Unit = { - val out01 = "from file:///appcom/bdap/johnnwang/ to hdfs:///tmp/bdp-ide/johnnwang/test1/new_py_180910175033.python" - val regex = ("(?i)\\s*from\\s+(\\S+)\\s+to\\s+(\\S+)\\s?").r - out01 match { - case regex(source,desct) => print(source.contains(".")) - } - + * created by patinousward on 2019/12/9 + * Description: + */ +class PipeEngineJob extends CommonEngineJob { + override def kill(): Unit = { + info("begin to remove osCache:" + this.getId) + OutputStreamCache.osCache.get(this.getId).close() //因为SingleTaskInfoSupport，应该是没有并发问题的 + super.kill() } - } diff --git a/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/PipeLineEngineExecutor.scala b/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/PipeLineEngineExecutor.scala index a0026aa03b..9e1c434484 100644 --- a/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/PipeLineEngineExecutor.scala +++ b/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/PipeLineEngineExecutor.scala @@ -16,7 +16,6 @@ package com.webank.wedatasphere.linkis.engine.pipeline - import com.webank.wedatasphere.linkis.common.utils.Logging import com.webank.wedatasphere.linkis.engine.execute.{EngineExecutor, EngineExecutorContext} import com.webank.wedatasphere.linkis.engine.pipeline.exception.PipeLineErrorException @@ -29,8 +28,8 @@ import com.webank.wedatasphere.linkis.scheduler.executer.{ExecuteResponse, Singl import com.webank.wedatasphere.linkis.server._ /** - * Created by johnnwang on 2018/11/13. - */ + * Created by johnnwang on 2018/11/13. + */ class PipeLineEngineExecutor(options: JMap[String, String]) extends EngineExecutor(outputPrintLimit = 10, false) with SingleTaskInfoSupport with Logging { override def getName: String = "pipeLineEngine" @@ -38,7 +37,7 @@ class PipeLineEngineExecutor(options: JMap[String, String]) extends EngineExecut private var index = 0 private var progressInfo: JobProgressInfo = _ - override def getActualUsedResources: Resource = new LoadInstanceResource(Runtime.getRuntime.totalMemory() - Runtime.getRuntime.freeMemory(), 2, 1) + override def getActualUsedResources: Resource = new LoadInstanceResource(Runtime.getRuntime.totalMemory() - Runtime.getRuntime.freeMemory(), 1, 1) override def init(): Unit = { info("init pipeLineEngine...") @@ -64,9 +63,9 @@ class PipeLineEngineExecutor(options: JMap[String, String]) extends EngineExecut code match { case regex(sourcePath, destPath) => { if (destPath.contains(".")) { - PipeLineExecutorFactory.listPipeLineExecutor.find(f => "cp".equals(f.Kind)).get.execute(sourcePath, destPath) + PipeLineExecutorFactory.listPipeLineExecutor.find(f => "cp".equals(f.Kind)).get.execute(sourcePath, destPath, engineExecutorContext) } else { - PipeLineExecutorFactory.listPipeLineExecutor.find(f => newOptions.get("pipeline.output.mold").equalsIgnoreCase(f.Kind)).map(_.execute(sourcePath, destPath)).get + PipeLineExecutorFactory.listPipeLineExecutor.find(f => newOptions.get("pipeline.output.mold").equalsIgnoreCase(f.Kind)).map(_.execute(sourcePath, destPath, engineExecutorContext)).get } } case _ => throw new PipeLineErrorException(70007, "") @@ -75,6 +74,8 @@ class PipeLineEngineExecutor(options: JMap[String, String]) extends EngineExecut case e: Exception => failedTasks = 1; succeedTasks = 0; throw e } finally { + info("begin to remove osCache:" + engineExecutorContext.getJobId.get) + OutputStreamCache.osCache.remove(engineExecutorContext.getJobId.get) progressInfo = JobProgressInfo(getName + "_" + index, 1, 0, failedTasks, succeedTasks) } } diff --git a/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/PipeLineEngineExecutorFactory.scala b/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/PipeLineEngineExecutorFactory.scala index 202a6dfeb4..d80df424a2 100644 --- a/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/PipeLineEngineExecutorFactory.scala +++ b/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/PipeLineEngineExecutorFactory.scala @@ -17,14 +17,12 @@ package com.webank.wedatasphere.linkis.engine.pipeline import com.webank.wedatasphere.linkis.engine.execute.{EngineExecutor, EngineExecutorFactory} -import com.webank.wedatasphere.linkis.engine.pipeline.exception.PipeLineErrorException -import com.webank.wedatasphere.linkis.engine.pipeline.executor.{CSVExecutor, CopyExecutor, ExcelExecutor} import com.webank.wedatasphere.linkis.server.JMap import org.springframework.stereotype.Component /** - * Created by johnnwang on 2018/11/13. - */ + * Created by johnnwang on 2018/11/13. + */ @Component("engineExecutorFactory") class PipeLineEngineExecutorFactory extends EngineExecutorFactory { override def createExecutor(options: JMap[String, String]): EngineExecutor = new PipeLineEngineExecutor(options) diff --git a/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/PipeTest.scala b/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/PipelineEngineParser.scala similarity index 55% rename from ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/PipeTest.scala rename to ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/PipelineEngineParser.scala index a354a4a65b..1d6cf258a1 100644 --- a/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/PipeTest.scala +++ b/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/PipelineEngineParser.scala @@ -13,23 +13,22 @@ * See the License for the specific language governing permissions and * limitations under the License. */ - package com.webank.wedatasphere.linkis.engine.pipeline -import com.webank.wedatasphere.linkis.engine.pipeline.domain.PipeEntity -import com.webank.wedatasphere.linkis.engine.pipeline.util.PipeLineUtil +import com.webank.wedatasphere.linkis.engine.EngineParser +import com.webank.wedatasphere.linkis.protocol.engine.RequestTask +import com.webank.wedatasphere.linkis.scheduler.queue.Job +import org.springframework.stereotype.Component /** - * Created by johnnwang on 2018/11/14. - */ -object PipeTest { - def main(args: Array[String]): Unit = { - val out01 = "from file:///appcom/bdap/johnnwang/ to hdfs:///tmp/bdp-ide/johnnwang/test1/new_py_180910175033.python" - val regex = ("(?i)\\s*from\\s+(\\S+)\\s+to\\s+(\\S+)\\s?").r - out01 match { - case regex(source,desct) => print(source.contains(".")) - } - + * created by patinousward on 2019/12/9 + * Description: + */ +@Component("engineParser") +class PipelineEngineParser extends EngineParser { + override def parseToJob(request: RequestTask): Job = { + val job = new PipeEngineJob + job.setRequestTask(request) + job } - } diff --git a/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/exception/PipeLineErrorException.scala b/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/constant/PipeLineConstant.scala similarity index 53% copy from ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/exception/PipeLineErrorException.scala copy to ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/constant/PipeLineConstant.scala index 2f3a8ec394..f06bfe935d 100644 --- a/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/exception/PipeLineErrorException.scala +++ b/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/constant/PipeLineConstant.scala @@ -14,13 +14,18 @@ * limitations under the License. */ -package com.webank.wedatasphere.linkis.engine.pipeline.exception - -import com.webank.wedatasphere.linkis.common.exception.ErrorException +package com.webank.wedatasphere.linkis.engine.pipeline.constant /** - * Created by johnnwang on 2018/11/16. + * Created by patinousward on 2018/11/16. */ -class PipeLineErrorException(errCode: Int, desc: String) extends ErrorException(errCode, desc) { - +object PipeLineConstant { + val DEFAULTC_HARSET = "utf-8" + val DEFAULT_SHEETNAME = "result" + val DEFAULT_DATEFORMATE = "yyyy-MM-dd HH:mm:ss" + val PIPELINE_OUTPUT_ISOVERWRITE = "pipeline.output.isoverwtite" + val PIPELINE_OUTPUT_SHUFFLE_NULL_TYPE = "pipeline.output.shuffle.null.type" + val PIPELINE_OUTPUT_CHARSET = "pipeline.output.charset" + val PIPELINE_FIELD_SPLIT = "pipeline.field.split" + val BLANK = "BLANK" } diff --git a/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/exception/PipeLineErrorException.scala b/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/conversions/FsConversions.scala similarity index 60% copy from ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/exception/PipeLineErrorException.scala copy to ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/conversions/FsConversions.scala index 2f3a8ec394..dddc624b8c 100644 --- a/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/exception/PipeLineErrorException.scala +++ b/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/conversions/FsConversions.scala @@ -14,13 +14,24 @@ * limitations under the License. */ -package com.webank.wedatasphere.linkis.engine.pipeline.exception +package com.webank.wedatasphere.linkis.engine.pipeline.conversions -import com.webank.wedatasphere.linkis.common.exception.ErrorException +import java.io.Closeable + +import com.webank.wedatasphere.linkis.common.io.Fs /** - * Created by johnnwang on 2018/11/16. - */ -class PipeLineErrorException(errCode: Int, desc: String) extends ErrorException(errCode, desc) { + * Created by patinousward on 2020/2/4. + */ +object FsConversions { + + implicit def fsToFsClosable(fs: Fs): Closeable = { + new FsClosable(fs) + } +} +class FsClosable(fs: Fs) extends Closeable { + override def close(): Unit = { + fs.close() + } } diff --git a/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/exception/PipeLineErrorException.scala b/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/exception/PipeLineErrorException.scala index 2f3a8ec394..9e4b409d03 100644 --- a/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/exception/PipeLineErrorException.scala +++ b/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/exception/PipeLineErrorException.scala @@ -19,8 +19,8 @@ package com.webank.wedatasphere.linkis.engine.pipeline.exception import com.webank.wedatasphere.linkis.common.exception.ErrorException /** - * Created by johnnwang on 2018/11/16. - */ + * Created by johnnwang on 2018/11/16. + */ class PipeLineErrorException(errCode: Int, desc: String) extends ErrorException(errCode, desc) { } diff --git a/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/executor/CSVExecutor.scala b/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/executor/CSVExecutor.scala index fd034988a9..4f9c594005 100644 --- a/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/executor/CSVExecutor.scala +++ b/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/executor/CSVExecutor.scala @@ -15,64 +15,61 @@ */ package com.webank.wedatasphere.linkis.engine.pipeline.executor -import java.io.{ByteArrayInputStream, ByteArrayOutputStream, InputStream, OutputStream} + +import java.io.OutputStream import com.webank.wedatasphere.linkis.common.io.FsPath import com.webank.wedatasphere.linkis.engine.execute.EngineExecutorContext +import com.webank.wedatasphere.linkis.engine.pipeline.OutputStreamCache +import com.webank.wedatasphere.linkis.engine.pipeline.constant.PipeLineConstant._ +import com.webank.wedatasphere.linkis.engine.pipeline.conversions.FsConversions._ import com.webank.wedatasphere.linkis.engine.pipeline.exception.PipeLineErrorException -import com.webank.wedatasphere.linkis.engine.pipeline.util.{PipeLineConstants, PipeLineUtils} -import com.webank.wedatasphere.linkis.scheduler.executer.{ExecuteResponse, SuccessExecuteResponse} +import com.webank.wedatasphere.linkis.scheduler.executer.ExecuteResponse import com.webank.wedatasphere.linkis.server._ import com.webank.wedatasphere.linkis.storage.FSFactory import com.webank.wedatasphere.linkis.storage.csv.CSVFsWriter -import com.webank.wedatasphere.linkis.storage.excel.ExcelFsWriter -import com.webank.wedatasphere.linkis.storage.resultset.{ResultSetFactory, ResultSetReader} +import com.webank.wedatasphere.linkis.storage.source.FileSource import org.apache.commons.io.IOUtils -import org.springframework.stereotype.Component /** - * Created by johnnwang on 2019/1/30. - */ + * Created by johnnwang on 2019/1/30. + */ -class CSVExecutor extends PipeLineExecutor{ +class CSVExecutor extends PipeLineExecutor { - override def execute(sourcePath: String, destPath: String): ExecuteResponse = { - //val sourcePath = pipeEntity.getSource - if (!PipeLineUtils.isDolphin(sourcePath)) throw new PipeLineErrorException(70005, "Not a result set file（不是结果集文件）") + override def execute(sourcePath: String, destPath: String, engineExecutorContext: EngineExecutorContext): ExecuteResponse = { + if (!FileSource.isResultSet(sourcePath)) { + throw new PipeLineErrorException(70005, "Not a result set file（不是结果集文件）") + } val sourceFsPath = new FsPath(sourcePath) - //val destPath = pipeEntity.getDest - val destFsPath = new FsPath(destPath+"."+Kind) + val destFsPath = new FsPath(s"$destPath.$Kind") val sourceFs = FSFactory.getFs(sourceFsPath) sourceFs.init(null) val destFs = FSFactory.getFs(destFsPath) destFs.init(null) - val resultset = ResultSetFactory.getInstance.getResultSetByPath(sourceFsPath) - val reader = ResultSetReader.getResultSetReader(resultset, sourceFs.read(sourceFsPath)) - val metadata = reader.getMetaData - if (!PipeLineUtils.isTableResultset(metadata)) throw new PipeLineErrorException(70005, "Only the result set of the table type can be converted to csv（只有table类型的结果集才能转为csv）"); - val cSVFsWriter = CSVFsWriter.getCSVFSWriter(options.get("pipeline.output.charset"),options.get("pipeline.field.split")) - cSVFsWriter.addMetaData(metadata) - while (reader.hasNext){ - cSVFsWriter.addRecord(reader.getRecord) + val fileSource = FileSource.create(sourceFsPath, sourceFs) + if (!FileSource.isTableResultSet(fileSource)) { + throw new PipeLineErrorException(70005, "只有table类型的结果集才能转为csv") } - val inputStream = cSVFsWriter.getCSVStream - val outputStream:OutputStream = destFs.write(destFsPath,options.get("pipeline.output.isoverwtite").toBoolean) - // TODO: a series of close(一系列的close) - IOUtils.copy(inputStream,outputStream) - IOUtils.closeQuietly(outputStream) - IOUtils.closeQuietly(inputStream) - if (cSVFsWriter != null) cSVFsWriter.close() - if (reader != null) reader.close() - if(sourceFs != null) sourceFs.close() - if(destFs != null) destFs.close() - cleanOptions - SuccessExecuteResponse() + var nullValue = options.getOrDefault(PIPELINE_OUTPUT_SHUFFLE_NULL_TYPE, "NULL") + if (BLANK.equalsIgnoreCase(nullValue)) nullValue = "" + val outputStream: OutputStream = destFs.write(destFsPath, options.get(PIPELINE_OUTPUT_ISOVERWRITE).toBoolean) + OutputStreamCache.osCache += engineExecutorContext.getJobId.get -> outputStream + val cSVFsWriter = CSVFsWriter.getCSVFSWriter(options.get(PIPELINE_OUTPUT_CHARSET), options.get(PIPELINE_FIELD_SPLIT), outputStream) + fileSource.addParams("nullValue", nullValue).write(cSVFsWriter) + IOUtils.closeQuietly(cSVFsWriter) + IOUtils.closeQuietly(fileSource) + IOUtils.closeQuietly(sourceFs) + IOUtils.closeQuietly(destFs) + super.execute(sourcePath, destPath, engineExecutorContext) } override def Kind: String = "csv" } -object CSVExecutor{ + +object CSVExecutor { val csvExecutor = new CSVExecutor - def getInstance:PipeLineExecutor = csvExecutor + + def getInstance: PipeLineExecutor = csvExecutor } diff --git a/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/executor/CopyExecutor.scala b/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/executor/CopyExecutor.scala index 2f103b8e3f..c6e321af56 100644 --- a/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/executor/CopyExecutor.scala +++ b/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/executor/CopyExecutor.scala @@ -15,52 +15,53 @@ */ package com.webank.wedatasphere.linkis.engine.pipeline.executor + import com.webank.wedatasphere.linkis.common.io.FsPath import com.webank.wedatasphere.linkis.engine.execute.EngineExecutorContext -import com.webank.wedatasphere.linkis.engine.pipeline.util.PipeLineConstants -import com.webank.wedatasphere.linkis.scheduler.executer.{ExecuteResponse, SuccessExecuteResponse} +import com.webank.wedatasphere.linkis.engine.pipeline.OutputStreamCache +import com.webank.wedatasphere.linkis.engine.pipeline.constant.PipeLineConstant._ +import com.webank.wedatasphere.linkis.engine.pipeline.conversions.FsConversions._ +import com.webank.wedatasphere.linkis.scheduler.executer.ExecuteResponse import com.webank.wedatasphere.linkis.server._ import com.webank.wedatasphere.linkis.storage.FSFactory import org.apache.commons.io.IOUtils -import org.springframework.stereotype.Component /** - * Created by johnnwang on 2019/1/30. - */ + * Created by johnnwang on 2019/1/30. + */ -class CopyExecutor extends PipeLineExecutor{ - override def execute(sourcePath: String, destPath: String): ExecuteResponse = { - //val sourcePath = pipeEntity.getSource +class CopyExecutor extends PipeLineExecutor { + override def execute(sourcePath: String, destPath: String, engineExecutorContext: EngineExecutorContext): ExecuteResponse = { val sourceFsPath = new FsPath(sourcePath) - //val destPath = pipeEntity.getDest val destFsPath = new FsPath(destPath) val sourceFs = FSFactory.getFs(sourceFsPath) sourceFs.init(null) val destFs = FSFactory.getFs(destFsPath) destFs.init(null) val inputStream = sourceFs.read(sourceFsPath) - var isOverWrite = options.get("pipeline.output.isoverwtite").toBoolean - //The export table is currently only exported to the workspace, so other places are temporarily not modified.（导出表目前因为是只导出到工作空间，所以别的地方暂时不修改） - if(!isOverWrite && !destFs.exists(destFsPath)){ + var isOverWrite = options.get(PIPELINE_OUTPUT_ISOVERWRITE).toBoolean + //导出表目前因为是只导出到工作空间，所以别的地方暂时不修改 + if (!isOverWrite && !destFs.exists(destFsPath)) { isOverWrite = true } val outputStream = destFs.write(destFsPath, isOverWrite) + OutputStreamCache.osCache += engineExecutorContext.getJobId.get -> outputStream IOUtils.copy(inputStream, outputStream) - // TODO: a series of close（一系列的close） - outputStream.close() - inputStream.close() - sourceFs.close() - destFs.close() - cleanOptions - SuccessExecuteResponse() + IOUtils.closeQuietly(outputStream) + IOUtils.closeQuietly(inputStream) + IOUtils.closeQuietly(sourceFs) + IOUtils.closeQuietly(destFs) + super.execute(sourcePath, destPath, engineExecutorContext) } override def Kind: String = "cp" } -object CopyExecutor{ +object CopyExecutor { val copyExecutor = new CopyExecutor - def getInstance:PipeLineExecutor = copyExecutor + + def getInstance: PipeLineExecutor = copyExecutor } + diff --git a/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/executor/ExcelExecutor.scala b/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/executor/ExcelExecutor.scala index 5894074b31..5adaca3114 100644 --- a/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/executor/ExcelExecutor.scala +++ b/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/executor/ExcelExecutor.scala @@ -15,62 +15,59 @@ */ package com.webank.wedatasphere.linkis.engine.pipeline.executor -import java.io.{ByteArrayInputStream, ByteArrayOutputStream, InputStream, OutputStream} + +import java.io.OutputStream import com.webank.wedatasphere.linkis.common.io.FsPath +import com.webank.wedatasphere.linkis.engine.execute.EngineExecutorContext +import com.webank.wedatasphere.linkis.engine.pipeline.OutputStreamCache +import com.webank.wedatasphere.linkis.engine.pipeline.constant.PipeLineConstant._ +import com.webank.wedatasphere.linkis.engine.pipeline.conversions.FsConversions._ import com.webank.wedatasphere.linkis.engine.pipeline.exception.PipeLineErrorException -import com.webank.wedatasphere.linkis.engine.pipeline.util.{PipeLineConstants, PipeLineUtils} -import com.webank.wedatasphere.linkis.scheduler.executer.{ExecuteResponse, SuccessExecuteResponse} +import com.webank.wedatasphere.linkis.scheduler.executer.ExecuteResponse import com.webank.wedatasphere.linkis.storage.FSFactory import com.webank.wedatasphere.linkis.storage.excel.ExcelFsWriter -import com.webank.wedatasphere.linkis.storage.resultset.{ResultSetFactory, ResultSetReader} +import com.webank.wedatasphere.linkis.storage.source.FileSource import org.apache.commons.io.IOUtils /** - * Created by johnnwang on 2019/1/30. - */ + * Created by johnnwang on 2019/1/30. + */ class ExcelExecutor extends PipeLineExecutor { - override def execute(sourcePath: String, destPath: String): ExecuteResponse = { - //val sourcePath = pipeEntity.getSource - if (!PipeLineUtils.isDolphin(sourcePath)) throw new PipeLineErrorException(70005, "Not a result set file（不是结果集文件）") + override def execute(sourcePath: String, destPath: String, engineExecutorContext: EngineExecutorContext): ExecuteResponse = { + if (!FileSource.isResultSet(sourcePath)) { + throw new PipeLineErrorException(70005, "不是结果集文件") + } val sourceFsPath = new FsPath(sourcePath) - // val destPath = pipeEntity.getDest - val destFsPath = new FsPath(destPath+".xlsx") + val destFsPath = new FsPath(s"$destPath.xlsx") val sourceFs = FSFactory.getFs(sourceFsPath) sourceFs.init(null) val destFs = FSFactory.getFs(destFsPath) destFs.init(null) - val resultset = ResultSetFactory.getInstance.getResultSetByPath(sourceFsPath) - val reader = ResultSetReader.getResultSetReader(resultset, sourceFs.read(sourceFsPath)) - val metadata = reader.getMetaData - if (!PipeLineUtils.isTableResultset(metadata)) throw new PipeLineErrorException(70005, "Only the result set of the table type can be converted to excel（只有table类型的结果集才能转为excel）") - val excelFsWriter = ExcelFsWriter.getExcelFsWriter(PipeLineConstants.DEFAULTCHARSET, PipeLineConstants.DEFAULTSHEETNAME, PipeLineConstants.DEFAULTDATEFORMATE) - excelFsWriter.addMetaData(metadata) - while (reader.hasNext){ - excelFsWriter.addRecord(reader.getRecord) + val fileSource = FileSource.create(sourceFsPath, sourceFs) + if (!FileSource.isTableResultSet(fileSource)) { + throw new PipeLineErrorException(70005, "只有table类型的结果集才能转为excel") } - val os: ByteArrayOutputStream = new ByteArrayOutputStream() - excelFsWriter.getWorkBook.write(os ) - val inputStream:InputStream = new ByteArrayInputStream(os.toByteArray) - val outputStream:OutputStream = destFs.write(destFsPath,options.get("pipeline.output.isoverwtite").toBoolean) - // TODO: a series of close(一系列的close) - IOUtils.copy(inputStream,outputStream) - IOUtils.closeQuietly(outputStream) - IOUtils.closeQuietly(inputStream) - IOUtils.closeQuietly(os) - if (excelFsWriter != null) excelFsWriter.close() - if (reader != null) reader.close() - if(sourceFs != null) sourceFs.close() - if(destFs != null) destFs.close() - cleanOptions - SuccessExecuteResponse() + var nullValue = options.getOrDefault(PIPELINE_OUTPUT_SHUFFLE_NULL_TYPE, "NULL") + if (BLANK.equalsIgnoreCase(nullValue)) nullValue = "" + val outputStream: OutputStream = destFs.write(destFsPath, options.get(PIPELINE_OUTPUT_ISOVERWRITE).toBoolean) + val excelFsWriter = ExcelFsWriter.getExcelFsWriter(DEFAULTC_HARSET, DEFAULT_SHEETNAME, DEFAULT_DATEFORMATE, outputStream) + import scala.collection.JavaConversions._ + OutputStreamCache.osCache += engineExecutorContext.getJobId.get -> outputStream + fileSource.addParams("nullValue", nullValue).write(excelFsWriter) + IOUtils.closeQuietly(excelFsWriter) + IOUtils.closeQuietly(fileSource) + IOUtils.closeQuietly(sourceFs) + IOUtils.closeQuietly(destFs) + super.execute(sourcePath, destPath, engineExecutorContext) } - override def Kind: String = "excel" + override def Kind: String = "excel" } -object ExcelExecutor{ - val excelExecutor = new ExcelExecutor - def getInstance:PipeLineExecutor = excelExecutor +object ExcelExecutor { + val excelExecutor = new ExcelExecutor + + def getInstance: PipeLineExecutor = excelExecutor } diff --git a/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/executor/PipeLineExecutor.scala b/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/executor/PipeLineExecutor.scala index ac085350f0..b570a118b7 100644 --- a/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/executor/PipeLineExecutor.scala +++ b/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/executor/PipeLineExecutor.scala @@ -17,19 +17,24 @@ package com.webank.wedatasphere.linkis.engine.pipeline.executor import com.webank.wedatasphere.linkis.engine.execute.EngineExecutorContext -import com.webank.wedatasphere.linkis.scheduler.executer.ExecuteResponse -import com.webank.wedatasphere.linkis.server._ +import com.webank.wedatasphere.linkis.scheduler.executer.{ExecuteResponse, SuccessExecuteResponse} /** - * Created by johnnwang on 2019/1/30. - */ + * Created by johnnwang on 2019/1/30. + */ trait PipeLineExecutor { var options: java.util.Map[String, String] = _ - def execute(sourcePath: String,destPath: String): ExecuteResponse - def Kind:String - def init(newOptions:java.util.Map[String, String]): Unit ={ + + def execute(sourcePath: String, destPath: String, engineExecutorContext: EngineExecutorContext): ExecuteResponse = { + cleanOptions + SuccessExecuteResponse() + } + + def Kind: String + + def init(newOptions: java.util.Map[String, String]): Unit = { options = newOptions } - def cleanOptions= options = null + def cleanOptions = options = null } diff --git a/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/executor/PipeLineExecutorFactory.scala b/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/executor/PipeLineExecutorFactory.scala index ec9a4329e0..debb93cebe 100644 --- a/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/executor/PipeLineExecutorFactory.scala +++ b/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/executor/PipeLineExecutorFactory.scala @@ -16,11 +16,9 @@ package com.webank.wedatasphere.linkis.engine.pipeline.executor -import org.springframework.beans.factory.annotation.Autowired - /** - * Created by johnnwang on 2019/1/30. - */ + * Created by johnnwang on 2019/1/30. + */ object PipeLineExecutorFactory { - def listPipeLineExecutor:Array[PipeLineExecutor] = Array(CopyExecutor.getInstance,CSVExecutor.getInstance,ExcelExecutor.getInstance) + def listPipeLineExecutor: Array[PipeLineExecutor] = Array(CopyExecutor.getInstance, CSVExecutor.getInstance, ExcelExecutor.getInstance) } diff --git a/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/executor/TxtExecutor.scala.bak b/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/executor/TxtExecutor.scala.bak deleted file mode 100644 index 868638aa0a..0000000000 --- a/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/executor/TxtExecutor.scala.bak +++ /dev/null @@ -1,48 +0,0 @@ -package com.webank.wedatasphere.linkis.engine.pipeline.executor -import java.io.{ByteArrayInputStream, InputStream, OutputStream} - -import com.webank.wedatasphere.linkis.common.io.FsPath -import com.webank.wedatasphere.linkis.engine.execute.EngineExecutorContext -import com.webank.wedatasphere.linkis.engine.pipeline.exception.PipeLineErrorException -import com.webank.wedatasphere.linkis.engine.pipeline.util.{PipeLineConstants, PipeLineUtils} -import com.webank.wedatasphere.linkis.scheduler.executer.{ExecuteResponse, SuccessExecuteResponse} -import com.webank.wedatasphere.linkis.server._ -import com.webank.wedatasphere.linkis.storage.{FSFactory, LineMetaData, LineRecord} -import com.webank.wedatasphere.linkis.storage.resultset.{ResultSetFactory, ResultSetReader} -import org.apache.commons.io.IOUtils - -/** - * Created by johnnwang on 2019/1/30. - */ -class TxtExecutor(options: JMap[String, String]) extends PipeLineExecutor{ - override def execute(code: String, engineExecutorContext: EngineExecutorContext, jobGroup: String): ExecuteResponse = { - val sourcePath = pipeEntity.getSource - if (!PipeLineUtils.isDolphin(sourcePath)) throw new PipeLineErrorException(70005, "不是结果集文件") - val sourceFsPath = new FsPath(sourcePath) - val destPath = pipeEntity.getDest - val destFsPath = new FsPath(destPath) - val sourceFs = FSFactory.getFs(sourceFsPath) - sourceFs.init(null) - val destFs = FSFactory.getFs(destFsPath) - destFs.init(null) - val resultset = ResultSetFactory.getInstance.getResultSetByPath(sourceFsPath) - val reader = ResultSetReader.getResultSetReader(resultset, sourceFs.read(sourceFsPath)) - val metadata = reader.getMetaData - if (!PipeLineUtils.isLineResultset(metadata)) throw new PipeLineErrorException(70006, "只有line类型的结果集才能转为txt"); - val content = new StringBuffer() - content.append(metadata.asInstanceOf[LineMetaData].getMetaData) - while(reader.hasNext){ - content.append(reader.getRecord.asInstanceOf[LineRecord].getLine) - } - val inputStream:InputStream = new ByteArrayInputStream(content.toString().getBytes(PipeLineConstants.DEFAULTCHARSET)) - val outputStream:OutputStream = destFs.write(destFsPath,PipeLineConstants.DEFAULTOVERWIRTE) - // TODO: a series of close（一系列的close） - IOUtils.copy(inputStream,outputStream) - IOUtils.closeQuietly(outputStream) - IOUtils.closeQuietly(inputStream) - if (reader != null) reader.close() - if(sourceFs != null) sourceFs.close() - if(destFs != null) destFs.close() - SuccessExecuteResponse() - } -} diff --git a/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/parser/IEParser.scala b/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/parser/IEParser.scala deleted file mode 100644 index cc33c18132..0000000000 --- a/ujes/definedEngines/pipeline/engine/src/main/scala/com/webank/wedatasphere/linkis/engine/pipeline/parser/IEParser.scala +++ /dev/null @@ -1,104 +0,0 @@ -/* - * Copyright 2019 WeBank - * - * 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 com.webank.wedatasphere.linkis.engine.pipeline.parser - -import com.webank.wedatasphere.linkis.common.utils.Utils -import com.webank.wedatasphere.linkis.engine.pipeline.exception.PipeLineErrorException -import org.apache.commons.lang.StringUtils - -/** - * Created by johnnwang on 2018/11/14. - * import and export parser - */ -object IEParser { - - val RESTART_CODE = "@restart" - - val APPLICATION_START_COMMAND = List("\\s*@restart\\s*", - "\\s*[@|%][a-zA-Z]{1,12}\\s*", - "^\\s*@set\\s*spark\\..+\\s*", - "^\\s*#.+\\s*", - "^\\s*//.+\\s*", - "^\\s*--.+\\s*", - "\\s*") - - def needToRestart(code: String): Boolean = code.startsWith(RESTART_CODE) - - private def getIndex(_code: String, start: Int): Int = { - val index1 = _code.indexOf("\n", start) - val index2 = _code.indexOf("\\n", start) - if (index1 > -1 && index2 > -1) Math.min(index1, index2) else Math.max(index1, index2) - } - - def getKindString(code: String): String = { - val _code = StringUtils.strip(code) - - var start = 0 - if (_code.startsWith(RESTART_CODE)) { - start = getIndex(_code, 0) + 1 - } - var index = getIndex(_code, start) - if (index == -1) index = _code.length - StringUtils.strip(_code.substring(start, index)) - } - - def getKind(code: String): String = { - val kindStr = getKindString(code) - if (kindStr.matches("[%|@][a-zA-Z]{1,12}[\\.][a-zA-Z]{1,12}")) kindStr.substring(1) else throw new PipeLineErrorException(70004,"unknown kind") - } - - - /** - * This method just removes @restart and language identifiers (such as %sql, %scala, etc.), that is, removes at most the first 2 rows. - * 该方法只是去掉了@restart和语言标识符（如%sql、%scala等），即最多只去掉最前面2行 - * - * @param code - * @return - */ - def getRealCode(code: String): String = { - val _code = StringUtils.strip(code) - val kindStr = getKindString(_code) - if (kindStr.matches("[%|@][a-zA-Z]{1,12}")) - StringUtils.strip(_code.substring(_code.indexOf(kindStr) + kindStr.length)) - else if (_code.startsWith(RESTART_CODE)) StringUtils.strip(_code.substring(RESTART_CODE.length)) - else _code - } - - /** - * This method removes all code-independent setting parameters and identifiers (including comments) - * 该方法去掉一切与代码无关的设置参数和标识符（包括注释） - * - * @param code - * @return - */ - def getFormatCode(code: String): String = { - val msg = new StringBuilder - val restartRegex = "\\s*@restart\\s*".r - val kindRegex = "\\s*[@|%][a-zA-Z]{1,12}\\s*".r - val setRegex = "^\\s*@set\\s*.+\\s*".r - val symbolRegex1 = "^\\s*#.+\\s*".r - val symbolRegex2 = "^\\s*//.+\\s*".r - val symbolRegex3 = "\\s*--.+\\s*".r - val blankRegex = "\\s*".r - code.split("\n").foreach { - case blankRegex() | setRegex() | symbolRegex1() | symbolRegex2() | symbolRegex3() | restartRegex() | kindRegex() => - case str => msg ++= str ++ "\n" - } - StringUtils.strip(msg.toString()) - } - -} --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
