Skip to content

Commit

Permalink
Update the pb and refactor (#16)
Browse files Browse the repository at this point in the history
* add "how to complie pb" to readme.md

* update sqlflow.proto and temporary saving

* refactor jsqlflow

* add test for detecting html

* add comments and reduce interface functions
  • Loading branch information
weiguoz authored Jan 7, 2020
1 parent 21742f8 commit 549f020
Show file tree
Hide file tree
Showing 9 changed files with 437 additions and 211 deletions.
12 changes: 11 additions & 1 deletion README.md
Original file line number Diff line number Diff line change
@@ -1,4 +1,14 @@
# SQLFlow client for Java Developers

## Requirements
- Java 8+
- Java 8+

## Build
### Compile protobuf
```shell script
mvn protobuf:compile && mvn protobuf:compile-custom
```
### Package
```shell script
mvn clean package -q
```
40 changes: 40 additions & 0 deletions src/main/java/org/sqlflow/client/MessageHandler.java
Original file line number Diff line number Diff line change
@@ -0,0 +1,40 @@
/*
* Copyright 2019 The SQLFlow Authors. All rights reserved.
* 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.sqlflow.client;

import java.util.List;

/** Interface for application callback objects to receive logs and messages from SQLFlow server. */
public interface MessageHandler {
/** Handle the html format response */
void handleHTML(String html);

/** Handle the log */
void handleText(String text);

/** EndOfExecution */
void handleEOE();

/** Table header. */
void handleHeader(List<String> columnNames);

/**
* Results
*
* <p>TODO(weiguo): shouldn't expose `Any`
*/
void handleRows(List<com.google.protobuf.Any> rows);
}
186 changes: 152 additions & 34 deletions src/main/java/org/sqlflow/client/SQLFlow.java
Original file line number Diff line number Diff line change
Expand Up @@ -15,41 +15,159 @@

package org.sqlflow.client;

import java.util.Iterator;
import java.util.List;
import java.util.NoSuchElementException;
import java.util.concurrent.TimeUnit;

import com.google.protobuf.Any;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.StatusRuntimeException;
import proto.Sqlflow.JobStatus;
import org.apache.commons.lang3.StringUtils;
import org.sqlflow.client.utils.HTMLDetector;
import proto.SQLFlowGrpc;
import proto.Sqlflow.FetchRequest;
import proto.Sqlflow.FetchResponse;
import proto.Sqlflow.FetchResponse.Logs;
import proto.Sqlflow.Job;
import proto.Sqlflow.Request;
import proto.Sqlflow.Response;
import proto.Sqlflow.Session;

public interface SQLFlow {
/**
* Submit a task to SQLFlow server. This method return immediately.
*
* @param session: specify dbConnStr(datasource), user Id ...
* mysql://root:root@tcp(localhost)/iris
* @param sql: sql program.
* <p>Example: "SELECT * FROM iris.test; SELECT * FROM iris.iris TO TRAIN DNNClassifier
* COLUMN..." *
* @return return a job id for tracking.
* @throws IllegalArgumentException header or sql error
* @throws StatusRuntimeException
*/
String submit(Session session, String sql)
throws IllegalArgumentException, StatusRuntimeException;

/**
* Fetch the job status by job id. The job id always returned by submit. By fetch(), we are able
* to tracking the job status
*
* @param jobId specific the job we are going to track
* @return see @code proto.JobStatus.Code
* @throws StatusRuntimeException
*/
JobStatus fetch(String jobId) throws StatusRuntimeException;

/**
* Close the opened channel to SQLFlow server. Waits for the channel to become terminated, giving
* up if the timeout is reached.
*
* @throws InterruptedException thrown by awaitTermination
*/
void release() throws InterruptedException;
public class SQLFlow {
private Builder builder;

private SQLFlowGrpc.SQLFlowBlockingStub blockingStub;
// TODO(weiguo): It looks we need the futureStub to handle a large data set.
// private SQLFlowGrpc.SQLFlowFutureStub futureStub;

private SQLFlow(Builder builder) {
this.builder = builder;
blockingStub = SQLFlowGrpc.newBlockingStub(builder.channel);
}

public void run(String sql)
throws IllegalArgumentException, StatusRuntimeException, NoSuchElementException {
if (StringUtils.isBlank(sql)) {
throw new IllegalArgumentException("sql is empty");
}

Request req = Request.newBuilder().setSession(builder.session).setSql(sql).build();
try {
Iterator<Response> responses = blockingStub.run(req);
handleSQLFlowResponses(responses);
} catch (StatusRuntimeException e) {
// TODO(weiguo) logger.error
throw e;
}
}

public void release() throws InterruptedException {
builder.channel.shutdown().awaitTermination(5, TimeUnit.SECONDS);
}

private void handleSQLFlowResponses(Iterator<Response> responses) {
if (responses == null || !responses.hasNext()) {
throw new NoSuchElementException("bad response");
}
while (responses.hasNext()) {
Response response = responses.next();
if (response == null) {
break;
}
if (response.hasHead()) {
builder.handler.handleHeader(response.getHead().getColumnNamesList());
} else if (response.hasRow()) {
List<Any> rows = response.getRow().getDataList();
builder.handler.handleRows(rows);
} else if (response.hasMessage()) {
String msg = response.getMessage().getMessage();
if (HTMLDetector.validate(msg)) {
builder.handler.handleHTML(msg);
} else {
builder.handler.handleText(msg);
}
} else if (response.hasEoe()) {
builder.handler.handleEOE();
// assert(!responses.hasNext())
} else if (response.hasJob()) {
trackingJobStatus(response.getJob().getId());
// assert(!responses.hasNext())
} else {
break;
}
}
}

private void trackingJobStatus(String jobId) {
Job job = Job.newBuilder().setId(jobId).build();
FetchRequest req = FetchRequest.newBuilder().setJob(job).build();
while (true) {
FetchResponse response = blockingStub.fetch(req);
Logs logs = response.getLogs();
logs.getContentList().forEach(this.builder.handler::handleText);
if (response.getEof()) {
this.builder.handler.handleEOE();
break;
}
req = response.getUpdatedFetchSince();

try {
Thread.sleep(builder.intervalFetching);
} catch (InterruptedException e) {
break;
}
}
}

public static class Builder {
private ManagedChannel channel;
private MessageHandler handler;
private Session session;
private long intervalFetching = 2000L; // millis

public static Builder newInstance() {
return new Builder();
}

public Builder withChannel(ManagedChannel channel) {
this.channel = channel;
return this;
}

public Builder withMessageHandler(MessageHandler handler) {
this.handler = handler;
return this;
}

public Builder withIntervalFetching(long mills) {
if (mills > 0) {
this.intervalFetching = mills;
}
return this;
}

public Builder withSession(Session session) {
if (session == null || StringUtils.isAnyBlank(session.getDbConnStr(), session.getUserId())) {
throw new IllegalArgumentException("data source and userId are not allowed to be empty");
}
this.session = session;
return this;
}

/**
* Open a channel to the SQLFlow server. The serverUrl argument always ends with a port.
*
* @param serverUrl an address the SQLFlow server exposed.
* <p>Example: "localhost:50051"
*/
public Builder forTarget(String serverUrl) {
return withChannel(ManagedChannelBuilder.forTarget(serverUrl).usePlaintext().build());
}

public SQLFlow build() {
return new SQLFlow(this);
}
}
}
103 changes: 0 additions & 103 deletions src/main/java/org/sqlflow/client/impl/SQLFlowImpl.java

This file was deleted.

42 changes: 42 additions & 0 deletions src/main/java/org/sqlflow/client/utils/HTMLDetector.java
Original file line number Diff line number Diff line change
@@ -0,0 +1,42 @@
/*
* Copyright 2019 The SQLFlow Authors. All rights reserved.
* 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.sqlflow.client.utils;

import java.util.regex.Pattern;

import org.apache.commons.lang3.StringUtils;

/**
* Test if a string contains HTML tag.
*
* <p>Thanks https://ideone.com/HakdHo
*/
public class HTMLDetector {
private static final String TAG_START =
"<\\w+((\\s+\\w+(\\s*=\\s*(?:\".*?\"|'.*?'|[^'\">\\s]+))?)+\\s*|\\s*)>";
private static final String TAG_END = "</\\w+>";
private static final String TAG_SELF_CLOSING =
"<\\w+((\\s+\\w+(\\s*=\\s*(?:\".*?\"|'.*?'|[^'\">\\s]+))?)+\\s*|\\s*)/>";
private static final String HTML_ENTITY = "&[a-zA-Z][a-zA-Z0-9]+;";
private static final Pattern HTML_PATTERN =
Pattern.compile(
"(" + TAG_START + ".*" + TAG_END + ")|(" + TAG_SELF_CLOSING + ")|(" + HTML_ENTITY + ")",
Pattern.DOTALL);

public static boolean validate(String str) {
return !StringUtils.isEmpty(str) && HTML_PATTERN.matcher(str).find();
}
}
Loading

0 comments on commit 549f020

Please sign in to comment.