-
Notifications
You must be signed in to change notification settings - Fork 710
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
126 changed files
with
4,012 additions
and
2,404 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file was deleted.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,12 +1,11 @@ | ||
|
|
||
| curl -XPUT 'localhost:9200/_river/my_jdbc_river/_meta' -d '{ | ||
| curl -XPUT '0:9200/_river/my_jdbc_river/_meta' -d '{ | ||
| "type" : "jdbc", | ||
| "jdbc" : { | ||
| "driver" : "com.mysql.jdbc.Driver", | ||
| "url" : "jdbc:mysql://localhost:3306/test", | ||
| "user" : "", | ||
| "password" : "", | ||
| "sql" : "select * from orders", | ||
| "strategy" : "simple" | ||
| "sql" : "select *, created as _id from orders", | ||
| "schedule" : "0/20 0-59 0-23 ? * *" | ||
| } | ||
| }' |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,10 @@ | ||
|
|
||
| curl -XPUT '0:9200/_river/my_jdbc_river/_meta' -d '{ | ||
| "type" : "jdbc", | ||
| "jdbc" : { | ||
| "url" : "jdbc:mysql://localhost:3306/test", | ||
| "user" : "", | ||
| "password" : "", | ||
| "sql" : "select *, created as _id from orders" | ||
| } | ||
| }' |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -5,7 +5,10 @@ | |
| import org.elasticsearch.plugins.AbstractPlugin; | ||
| import org.elasticsearch.rest.RestModule; | ||
| import org.elasticsearch.river.RiversModule; | ||
|
|
||
| import org.xbib.elasticsearch.rest.action.RestJDBCRiverCreateAction; | ||
| import org.xbib.elasticsearch.rest.action.RestJDBCRiverInduceAction; | ||
| import org.xbib.elasticsearch.rest.action.RestJDBCRiverStateAction; | ||
| import org.xbib.elasticsearch.river.jdbc.JDBCRiver; | ||
| import org.xbib.elasticsearch.river.jdbc.JDBCRiverModule; | ||
|
|
||
|
|
@@ -16,8 +19,6 @@ | |
| * binds a JDBCRiver instance to a River interface - and the | ||
| * REST move is configured. Also, a concurrent bulk move | ||
| * is registered. | ||
| * | ||
| * @author Jörg Prante <[email protected]> | ||
| */ | ||
| public class JDBCRiverPlugin extends AbstractPlugin { | ||
|
|
||
|
|
@@ -36,21 +37,23 @@ public String description() { | |
| } | ||
|
|
||
| /** | ||
| * Register the JDBC river to Elasticsearch node | ||
| * Register the river to Elasticsearch | ||
| * | ||
| * @param module | ||
| * @param module the rivers module | ||
| */ | ||
| public void onModule(RiversModule module) { | ||
| module.registerRiver(JDBCRiver.TYPE, JDBCRiverModule.class); | ||
| } | ||
|
|
||
| /** | ||
| * Register the REST move to Elasticsearch node | ||
| * Register the REST actions of this river | ||
| * | ||
| * @param module | ||
| * @param module the REST module | ||
| */ | ||
| public void onModule(RestModule module) { | ||
| module.addRestAction(RestJDBCRiverCreateAction.class); | ||
| module.addRestAction(RestJDBCRiverInduceAction.class); | ||
| module.addRestAction(RestJDBCRiverStateAction.class); | ||
| } | ||
|
|
||
| } | ||
72 changes: 72 additions & 0 deletions
72
src/main/java/org/xbib/elasticsearch/rest/action/AbstractRestRiverAction.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,72 @@ | ||
| package org.xbib.elasticsearch.rest.action; | ||
|
|
||
| import java.io.IOException; | ||
| import java.lang.reflect.Field; | ||
|
|
||
| import org.elasticsearch.client.Client; | ||
| import org.elasticsearch.common.collect.ImmutableMap; | ||
| import org.elasticsearch.common.inject.Injector; | ||
| import org.elasticsearch.common.settings.Settings; | ||
| import org.elasticsearch.common.xcontent.XContentBuilder; | ||
| import org.elasticsearch.rest.BaseRestHandler; | ||
| import org.elasticsearch.rest.RestChannel; | ||
| import org.elasticsearch.rest.RestRequest; | ||
| import org.elasticsearch.rest.RestStatus; | ||
| import org.elasticsearch.rest.XContentRestResponse; | ||
| import org.elasticsearch.rest.XContentThrowableRestResponse; | ||
| import org.elasticsearch.rest.action.support.RestXContentBuilder; | ||
| import org.elasticsearch.river.River; | ||
| import org.elasticsearch.river.RiverName; | ||
| import org.elasticsearch.river.RiversService; | ||
|
|
||
| public abstract class AbstractRestRiverAction extends BaseRestHandler { | ||
|
|
||
| protected Injector injector; | ||
|
|
||
| public AbstractRestRiverAction(Settings settings, Client client, Injector injector) { | ||
| super(settings, client); | ||
| this.injector = injector; | ||
| } | ||
|
|
||
| protected void respond(boolean success, RestRequest request, RestChannel channel, String error, RestStatus status) { | ||
| try { | ||
| XContentBuilder builder = RestXContentBuilder.restContentBuilder(request); | ||
| builder.startObject(); | ||
| builder.field("success", success); | ||
| if (error != null) { | ||
| builder.field("error", error); | ||
| } | ||
| builder.endObject(); | ||
| channel.sendResponse(new XContentRestResponse(request, status, builder)); | ||
| } catch (IOException e) { | ||
| errorResponse(request, channel, e); | ||
| } | ||
| } | ||
|
|
||
| protected void errorResponse(RestRequest request, RestChannel channel, Throwable e) { | ||
| try { | ||
| channel.sendResponse(new XContentThrowableRestResponse(request, e)); | ||
| } catch (IOException e1) { | ||
| logger.error("Failed to send failure response", e1); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Retrieve the registered rivers using reflection (UGLY HACK!!) | ||
| * TODO: Obtain the rivers using public API | ||
| * @param injector injector | ||
| * @return map of rivers or null if not possible | ||
| */ | ||
| protected ImmutableMap<RiverName, River> rivers(Injector injector) { | ||
| RiversService riversService = injector.getInstance(RiversService.class); | ||
| try { | ||
| Field field = RiversService.class.getDeclaredField("rivers"); | ||
| field.setAccessible(true); | ||
| return (ImmutableMap<RiverName, River>) field.get(riversService); | ||
| } catch (NoSuchFieldException e) { | ||
| return null; | ||
| } catch (IllegalAccessException e) { | ||
| return null; | ||
| } | ||
| } | ||
| } |
59 changes: 59 additions & 0 deletions
59
src/main/java/org/xbib/elasticsearch/rest/action/RestJDBCRiverCreateAction.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,59 @@ | ||
| package org.xbib.elasticsearch.rest.action; | ||
|
|
||
| import java.io.IOException; | ||
|
|
||
| import org.elasticsearch.client.Client; | ||
| import org.elasticsearch.common.inject.Inject; | ||
| import org.elasticsearch.common.inject.Injector; | ||
| import org.elasticsearch.common.settings.ImmutableSettings; | ||
| import org.elasticsearch.common.settings.Settings; | ||
| import org.elasticsearch.common.xcontent.XContentBuilder; | ||
| import org.elasticsearch.common.xcontent.XContentHelper; | ||
| import org.elasticsearch.rest.RestChannel; | ||
| import org.elasticsearch.rest.RestController; | ||
| import org.elasticsearch.rest.RestRequest; | ||
| import org.elasticsearch.rest.XContentRestResponse; | ||
| import org.elasticsearch.rest.XContentThrowableRestResponse; | ||
| import org.elasticsearch.river.RiverName; | ||
| import org.elasticsearch.river.RiverSettings; | ||
|
|
||
| import org.xbib.elasticsearch.river.jdbc.JDBCRiver; | ||
|
|
||
| import static org.elasticsearch.rest.RestStatus.OK; | ||
| import static org.elasticsearch.rest.action.support.RestXContentBuilder.restContentBuilder; | ||
|
|
||
| public class RestJDBCRiverCreateAction extends AbstractRestRiverAction { | ||
|
|
||
| private Settings settings = ImmutableSettings.EMPTY; | ||
|
|
||
| @Inject | ||
| public RestJDBCRiverCreateAction(Settings settings, Client client, | ||
| RestController controller, Injector injector) { | ||
| super(settings, client, injector); | ||
| this.settings = settings; | ||
| controller.registerHandler(RestRequest.Method.POST, "/_river/jdbc/{river}/create", this); | ||
| } | ||
|
|
||
| @Override | ||
| public void handleRequest(RestRequest request, RestChannel channel) { | ||
| try { | ||
| String riverName = request.param("river"); | ||
| RiverSettings riverSettings = new RiverSettings(settings, | ||
| XContentHelper.convertToMap(request.content(), true).v2()) ; | ||
| JDBCRiver river = new JDBCRiver(new RiverName("jdbc", riverName), riverSettings, client); | ||
| river.start(); | ||
| XContentBuilder builder = restContentBuilder(request); | ||
| builder.startObject() | ||
| .field("active", river.riverFlow().isActive()) | ||
| .endObject(); | ||
| channel.sendResponse(new XContentRestResponse(request, OK, builder)); | ||
| } catch (IOException ioe) { | ||
| try { | ||
| channel.sendResponse(new XContentThrowableRestResponse(request, ioe)); | ||
| } catch (IOException e) { | ||
| logger.error("unable to send response to client"); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| } |
Oops, something went wrong.