20 Commits

Author SHA1 Message Date
fe2eaa0588 v1.1.0 model poms 2025-03-05 16:21:14 -05:00
68471be9ef Merge branch 'develop' into stable 2025-03-05 16:18:49 -05:00
493f1f813d added reconcile/reindex/retry/fix/purge services 2025-03-05 13:28:41 -05:00
0ed41a39e4 fixed ASIE shard model parsing 2025-02-28 17:48:43 -05:00
3cd8c91f93 create ApiService and use it 2025-02-28 17:47:43 -05:00
0cb566e18d v1.3.x 2025-02-28 17:46:04 -05:00
c38ed7a73a minor POM typo 2025-02-28 17:07:23 -05:00
bf9a5fca50 add ASIE unit tests 2025-02-28 17:06:56 -05:00
40d13ac266 fix ASIE response model 2025-02-28 17:06:43 -05:00
75e25577b7 fix Solr response model 2025-02-28 16:58:36 -05:00
82410805db v1.2.2 poms 2025-01-09 16:08:48 -05:00
ceb8d2c690 Merge branch 'develop' into stable 2025-01-09 16:05:18 -05:00
35bae4283d get authorities from AuthorityService 2025-01-09 11:53:48 -05:00
d537c8ec49 logging authority for debugging 2025-01-09 11:05:46 -05:00
f17556835a fix afterPropertiesSet() 2025-01-08 17:10:33 -05:00
4531c7af8e changed admin to user auth; using configurable auth 2025-01-08 16:51:47 -05:00
692410f535 moved ASIE custom authorization to AbstractWebScript 2025-01-08 16:47:35 -05:00
1230a07a5a added transaction wrapper to REST declaration 2025-01-08 14:52:34 -05:00
47835d852f wrapped attributeService in tx 2025-01-08 14:33:14 -05:00
7535475581 refactored PersistedNode for serialization 2025-01-08 13:52:58 -05:00
105 changed files with 3549 additions and 603 deletions

View File

@@ -6,20 +6,20 @@
<parent>
<groupId>com.inteligr8.alfresco</groupId>
<artifactId>asie-platform-module-parent</artifactId>
<version>1.2.1</version>
<version>1.2.2</version>
<relativePath>../</relativePath>
</parent>
<groupId>com.inteligr8.alfresco</groupId>
<artifactId>asie-api</artifactId>
<version>1.0.0-asie2</version>
<version>1.1.0-asie2</version>
<packaging>jar</packaging>
<name>ASIE JAX-RS API</name>
<description>Alfresco Search &amp; Insight Engine JAX-RS API</description>
<name>ASIE Jakarta RS API</name>
<description>Alfresco Search &amp; Insight Engine Jakarta RS API</description>
<properties>
<alfresco.platform.version>6.0.0</alfresco.platform.version>
<alfresco.platform.version>23.2.0</alfresco.platform.version>
</properties>
<dependencyManagement>
@@ -38,11 +38,45 @@
<dependency>
<groupId>com.inteligr8</groupId>
<artifactId>solr-api</artifactId>
<version>1.0.0-solr6</version>
<version>1.1.0-solr6</version>
</dependency>
<dependency>
<groupId>org.alfresco</groupId>
<artifactId>alfresco-repository</artifactId>
<artifactId>alfresco-data-model</artifactId>
<exclusions>
<exclusion>
<groupId>*</groupId>
<artifactId>*</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-lang3</artifactId>
<version>3.17.0</version>
</dependency>
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-slf4j2-impl</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>com.inteligr8</groupId>
<artifactId>common-rest-client</artifactId>
<version>3.0.2-jersey</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.glassfish.jersey.inject</groupId>
<artifactId>jersey-hk2</artifactId>
<version>3.1.10</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.junit.jupiter</groupId>
<artifactId>junit-jupiter-api</artifactId>
<version>5.11.2</version>
<scope>test</scope>
</dependency>
</dependencies>

View File

@@ -1,12 +1,12 @@
package com.inteligr8.alfresco.asie.api;
import com.inteligr8.alfresco.asie.model.ActionResponse;
import com.inteligr8.alfresco.asie.model.ActionCoreResponse;
import com.inteligr8.alfresco.asie.model.EmptyResponse;
import com.inteligr8.alfresco.asie.model.core.CheckRequest;
import com.inteligr8.alfresco.asie.model.core.DisableIndexingRequest;
import com.inteligr8.alfresco.asie.model.core.EnableIndexingRequest;
import com.inteligr8.alfresco.asie.model.core.FixAction;
import com.inteligr8.alfresco.asie.model.core.FixRequest;
import com.inteligr8.alfresco.asie.model.core.FixResponseAction;
import com.inteligr8.alfresco.asie.model.core.IndexingStatusAction;
import com.inteligr8.alfresco.asie.model.core.NewCoreRequest;
import com.inteligr8.alfresco.asie.model.core.NewDefaultIndexRequest;
@@ -14,14 +14,15 @@ import com.inteligr8.alfresco.asie.model.core.PurgeRequest;
import com.inteligr8.alfresco.asie.model.core.ReindexRequest;
import com.inteligr8.alfresco.asie.model.core.ReportRequest;
import com.inteligr8.alfresco.asie.model.core.ReportResponse;
import com.inteligr8.alfresco.asie.model.core.RetryAction;
import com.inteligr8.alfresco.asie.model.core.RetryRequest;
import com.inteligr8.alfresco.asie.model.core.RetryResponseAction;
import com.inteligr8.alfresco.asie.model.core.SummaryRequest;
import com.inteligr8.alfresco.asie.model.core.SummaryResponse;
import com.inteligr8.alfresco.asie.model.core.UpdateCoreRequest;
import com.inteligr8.alfresco.asie.model.core.UpdateLog4jRequest;
import com.inteligr8.alfresco.asie.model.core.UpdateSharedRequest;
import com.inteligr8.solr.model.ResponseAction;
import com.inteligr8.solr.model.Action;
import com.inteligr8.solr.model.ActionResponse;
import jakarta.ws.rs.BeanParam;
import jakarta.ws.rs.GET;
@@ -42,43 +43,43 @@ public interface CoreAdminApi extends com.inteligr8.solr.api.CoreAdminApi {
@GET
@Produces(MediaType.APPLICATION_JSON)
ActionResponse<ResponseAction> updateCore(@BeanParam UpdateCoreRequest request);
ActionResponse<Action> updateCore(@BeanParam UpdateCoreRequest request);
@GET
@Produces(MediaType.APPLICATION_JSON)
ActionResponse<ResponseAction> check(@BeanParam CheckRequest request);
ActionResponse<Action> check(@BeanParam CheckRequest request);
@GET
@Produces(MediaType.APPLICATION_JSON)
ActionResponse<ResponseAction> updateShared(@BeanParam UpdateSharedRequest request);
ActionResponse<Action> updateShared(@BeanParam UpdateSharedRequest request);
@GET
@Produces(MediaType.APPLICATION_JSON)
ActionResponse<ResponseAction> updateLog4j(@BeanParam UpdateLog4jRequest request);
ActionResponse<Action> updateLog4j(@BeanParam UpdateLog4jRequest request);
@GET
@Produces(MediaType.APPLICATION_JSON)
ActionResponse<ResponseAction> purge(@BeanParam PurgeRequest request);
ActionCoreResponse<Action> purge(@BeanParam PurgeRequest request);
@GET
@Produces(MediaType.APPLICATION_JSON)
ActionResponse<ResponseAction> reindex(@BeanParam ReindexRequest request);
ActionCoreResponse<Action> reindex(@BeanParam ReindexRequest request);
@GET
@Produces(MediaType.APPLICATION_JSON)
ActionResponse<RetryResponseAction> retry(@BeanParam RetryRequest request);
ActionCoreResponse<RetryAction> retry(@BeanParam RetryRequest request);
@GET
@Produces(MediaType.APPLICATION_JSON)
ActionResponse<FixResponseAction> fix(@BeanParam FixRequest request);
ActionCoreResponse<FixAction> fix(@BeanParam FixRequest request);
@GET
@Produces(MediaType.APPLICATION_JSON)
ActionResponse<IndexingStatusAction> enableIndexing(@BeanParam EnableIndexingRequest request);
ActionCoreResponse<IndexingStatusAction> enableIndexing(@BeanParam EnableIndexingRequest request);
@GET
@Produces(MediaType.APPLICATION_JSON)
ActionResponse<IndexingStatusAction> disableIndexing(@BeanParam DisableIndexingRequest request);
ActionCoreResponse<IndexingStatusAction> disableIndexing(@BeanParam DisableIndexingRequest request);
@GET
@Produces(MediaType.APPLICATION_JSON)

View File

@@ -0,0 +1,18 @@
package com.inteligr8.alfresco.asie.model;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.inteligr8.solr.model.Action;
import com.inteligr8.solr.model.Cores;
@JsonIgnoreProperties(ignoreUnknown = true)
public class ActionCoreResponse<T extends Action> extends BaseResponse {
@JsonProperty(value = "action")
private Cores<T> cores;
public Cores<T> getCores() {
return cores;
}
}

View File

@@ -1,22 +0,0 @@
package com.inteligr8.alfresco.asie.model;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.annotation.JsonProperty.Access;
import com.inteligr8.solr.model.ResponseAction;
@JsonIgnoreProperties(ignoreUnknown = true)
public class ActionResponse<T extends ResponseAction> extends BaseResponse {
@JsonProperty(access = Access.READ_ONLY)
private T action;
public T getAction() {
return action;
}
protected void setAction(T action) {
this.action = action;
}
}

View File

@@ -1,41 +1,28 @@
package com.inteligr8.alfresco.asie.model;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.annotation.JsonProperty.Access;
public class BaseResponse extends com.inteligr8.solr.model.BaseResponse {
@JsonProperty(value = "STATUS", access = Access.READ_ONLY)
@JsonProperty(value = "STATUS")
private String reason;
@JsonProperty(value = "exception", access = Access.READ_ONLY)
@JsonProperty(value = "exception")
private String exception;
@JsonProperty(value = "msg", access = Access.READ_ONLY)
@JsonProperty(value = "msg")
private String message;
public String getReason() {
return reason;
}
protected void setReason(String reason) {
this.reason = reason;
}
public String getException() {
return exception;
}
protected void setException(String exception) {
this.exception = exception;
}
public String getMessage() {
return message;
}
protected void setMessage(String message) {
this.message = message;
}
}

View File

@@ -0,0 +1,25 @@
package com.inteligr8.alfresco.asie.model.core;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.inteligr8.solr.model.Action;
import com.inteligr8.solr.model.TransactionStatus;
@JsonIgnoreProperties(ignoreUnknown = true)
public class FixAction extends Action {
@JsonProperty(value = "txToReindex")
private TransactionStatus transactionStatus;
@JsonProperty(value = "aclChangeSetToReindex")
private TransactionStatus aclStatus;
public TransactionStatus getTransactionStatus() {
return transactionStatus;
}
public TransactionStatus getAclStatus() {
return aclStatus;
}
}

View File

@@ -1,34 +0,0 @@
package com.inteligr8.alfresco.asie.model.core;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.annotation.JsonProperty.Access;
import com.inteligr8.solr.model.ResponseAction;
import com.inteligr8.solr.model.TransactionResponseStatus;
@JsonIgnoreProperties(ignoreUnknown = true)
public class FixResponseAction extends ResponseAction {
@JsonProperty(value = "txToReindex", access = Access.READ_ONLY)
private TransactionResponseStatus transactionStatus;
@JsonProperty(value = "aclChangeSetToReindex", access = Access.READ_ONLY)
private TransactionResponseStatus aclStatus;
public TransactionResponseStatus getTransactionStatus() {
return transactionStatus;
}
protected void setTransactionStatus(TransactionResponseStatus transactionStatus) {
this.transactionStatus = transactionStatus;
}
public TransactionResponseStatus getAclStatus() {
return aclStatus;
}
protected void setAclStatus(TransactionResponseStatus aclStatus) {
this.aclStatus = aclStatus;
}
}

View File

@@ -1,25 +1,43 @@
package com.inteligr8.alfresco.asie.model.core;
import java.util.Map;
import com.fasterxml.jackson.annotation.JsonAnyGetter;
import com.fasterxml.jackson.annotation.JsonAnySetter;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.inteligr8.solr.model.ResponseAction;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.inteligr8.solr.model.Action;
@JsonIgnoreProperties(ignoreUnknown = true)
public class IndexingStatusAction extends ResponseAction {
public class IndexingStatusAction extends Action {
private Map<String, IndexingStatusMetadata> cores;
@JsonAnyGetter
public Map<String, IndexingStatusMetadata> getCores() {
return cores;
@JsonProperty(value = "ACL")
private Boolean aclIndexed;
@JsonProperty(value = "CONTENT")
private Boolean contentIndexed;
@JsonProperty(value = "METADATA")
private Boolean metadataIndexed;
public boolean isAclIndexed() {
return Boolean.TRUE.equals(this.aclIndexed);
}
@JsonAnySetter
public void setCores(Map<String, IndexingStatusMetadata> cores) {
this.cores = cores;
public Boolean getAclIndexed() {
return aclIndexed;
}
public boolean isContentIndexed() {
return Boolean.TRUE.equals(this.contentIndexed);
}
public Boolean getContentIndexed() {
return contentIndexed;
}
public boolean isMetadataIndexed() {
return Boolean.TRUE.equals(this.metadataIndexed);
}
public Boolean getMetadataIndexed() {
return metadataIndexed;
}
}

View File

@@ -1,56 +0,0 @@
package com.inteligr8.alfresco.asie.model.core;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.annotation.JsonProperty.Access;
import com.inteligr8.solr.model.ResponseAction;
@JsonIgnoreProperties(ignoreUnknown = true)
public class IndexingStatusMetadata extends ResponseAction {
@JsonProperty(value = "ACL", access = Access.READ_ONLY)
private Boolean aclIndexed;
@JsonProperty(value = "CONTENT", access = Access.READ_ONLY)
private Boolean contentIndexed;
@JsonProperty(value = "METADATA", access = Access.READ_ONLY)
private Boolean metadataIndexed;
public boolean isAclIndexed() {
return Boolean.TRUE.equals(this.aclIndexed);
}
public Boolean getAclIndexed() {
return aclIndexed;
}
protected void setAclIndexed(Boolean aclIndexed) {
this.aclIndexed = aclIndexed;
}
public boolean isContentIndexed() {
return Boolean.TRUE.equals(this.contentIndexed);
}
public Boolean getContentIndexed() {
return contentIndexed;
}
protected void setContentIndexed(Boolean contentIndexed) {
this.contentIndexed = contentIndexed;
}
public boolean isMetadataIndexed() {
return Boolean.TRUE.equals(this.metadataIndexed);
}
public Boolean getMetadataIndexed() {
return metadataIndexed;
}
protected void setMetadataIndexed(Boolean metadataIndexed) {
this.metadataIndexed = metadataIndexed;
}
}

View File

@@ -19,16 +19,16 @@ public class PurgeRequest extends JsonFormattedResponseRequest<PurgeRequest> {
private String core;
@QueryParam("txid")
private Integer transactionId;
private Long transactionId;
@QueryParam("acltxid")
private Integer aclTransactionId;
private Long aclTransactionId;
@QueryParam("nodeId")
private Integer nodeId;
private Long nodeId;
@QueryParam("aclid")
private Integer aclId;
private Long aclId;
public String getAction() {
return action;
@@ -51,54 +51,54 @@ public class PurgeRequest extends JsonFormattedResponseRequest<PurgeRequest> {
return this;
}
public Integer getTransactionId() {
public Long getTransactionId() {
return transactionId;
}
public void setTransactionId(Integer transactionId) {
public void setTransactionId(Long transactionId) {
this.transactionId = transactionId;
}
public PurgeRequest withTransactionId(Integer transactionId) {
public PurgeRequest withTransactionId(Long transactionId) {
this.transactionId = transactionId;
return this;
}
public Integer getAclTransactionId() {
public Long getAclTransactionId() {
return aclTransactionId;
}
public void setAclTransactionId(Integer aclTransactionId) {
public void setAclTransactionId(Long aclTransactionId) {
this.aclTransactionId = aclTransactionId;
}
public PurgeRequest withAclTransactionId(Integer aclTransactionId) {
public PurgeRequest withAclTransactionId(Long aclTransactionId) {
this.aclTransactionId = aclTransactionId;
return this;
}
public Integer getNodeId() {
public Long getNodeId() {
return nodeId;
}
public void setNodeId(Integer nodeId) {
public void setNodeId(Long nodeId) {
this.nodeId = nodeId;
}
public PurgeRequest withNodeId(Integer nodeId) {
public PurgeRequest withNodeId(Long nodeId) {
this.nodeId = nodeId;
return this;
}
public Integer getAclId() {
public Long getAclId() {
return aclId;
}
public void setAclId(Integer aclId) {
public void setAclId(Long aclId) {
this.aclId = aclId;
}
public PurgeRequest withAclId(Integer aclId) {
public PurgeRequest withAclId(Long aclId) {
this.aclId = aclId;
return this;
}

View File

@@ -19,16 +19,16 @@ public class ReindexRequest extends JsonFormattedResponseRequest<ReindexRequest>
private String core;
@QueryParam("txid")
private Integer transactionId;
private Long transactionId;
@QueryParam("acltxid")
private Integer aclTransactionId;
private Long aclTransactionId;
@QueryParam("nodeId")
private Integer nodeId;
private Long nodeId;
@QueryParam("aclid")
private Integer aclId;
private Long aclId;
@QueryParam("query")
private String query;
@@ -54,54 +54,54 @@ public class ReindexRequest extends JsonFormattedResponseRequest<ReindexRequest>
return this;
}
public Integer getTransactionId() {
public Long getTransactionId() {
return transactionId;
}
public void setTransactionId(Integer transactionId) {
public void setTransactionId(Long transactionId) {
this.transactionId = transactionId;
}
public ReindexRequest withTransactionId(Integer transactionId) {
public ReindexRequest withTransactionId(Long transactionId) {
this.transactionId = transactionId;
return this;
}
public Integer getAclTransactionId() {
public Long getAclTransactionId() {
return aclTransactionId;
}
public void setAclTransactionId(Integer aclTransactionId) {
public void setAclTransactionId(Long aclTransactionId) {
this.aclTransactionId = aclTransactionId;
}
public ReindexRequest withAclTransactionId(Integer aclTransactionId) {
public ReindexRequest withAclTransactionId(Long aclTransactionId) {
this.aclTransactionId = aclTransactionId;
return this;
}
public Integer getNodeId() {
public Long getNodeId() {
return nodeId;
}
public void setNodeId(Integer nodeId) {
public void setNodeId(Long nodeId) {
this.nodeId = nodeId;
}
public ReindexRequest withNodeId(Integer nodeId) {
public ReindexRequest withNodeId(Long nodeId) {
this.nodeId = nodeId;
return this;
}
public Integer getAclId() {
public Long getAclId() {
return aclId;
}
public void setAclId(Integer aclId) {
public void setAclId(Long aclId) {
this.aclId = aclId;
}
public ReindexRequest withAclId(Integer aclId) {
public ReindexRequest withAclId(Long aclId) {
this.aclId = aclId;
return this;
}

View File

@@ -1,24 +1,9 @@
package com.inteligr8.alfresco.asie.model.core;
import java.util.Map;
import com.fasterxml.jackson.annotation.JsonAnyGetter;
import com.fasterxml.jackson.annotation.JsonAnySetter;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.inteligr8.solr.model.Metadata;
@JsonIgnoreProperties(ignoreUnknown = true)
public class Report {
private Map<String, Map<String, Object>> report;
@JsonAnyGetter
public Map<String, Map<String, Object>> getReport() {
return report;
}
@JsonAnySetter
protected void setReport(Map<String, Map<String, Object>> report) {
this.report = report;
}
public class Report extends Metadata {
}

View File

@@ -25,10 +25,10 @@ public class ReportRequest extends JsonFormattedResponseRequest<ReportRequest> {
private Long toTime;
@QueryParam("fromTx")
private Integer fromTransactionId;
private Long fromTransactionId;
@QueryParam("toTx")
private Integer toTransactionId;
private Long toTransactionId;
public String getAction() {
return action;
@@ -77,28 +77,28 @@ public class ReportRequest extends JsonFormattedResponseRequest<ReportRequest> {
return this;
}
public Integer getFromTransactionId() {
public Long getFromTransactionId() {
return fromTransactionId;
}
public void setFromTransactionId(Integer fromTransactionId) {
public void setFromTransactionId(Long fromTransactionId) {
this.fromTransactionId = fromTransactionId;
}
public ReportRequest fromTransactionId(Integer fromTransactionId) {
public ReportRequest fromTransactionId(Long fromTransactionId) {
this.fromTransactionId = fromTransactionId;
return this;
}
public Integer getToTransactionId() {
public Long getToTransactionId() {
return toTransactionId;
}
public void setToTransactionId(Integer toTransactionId) {
public void setToTransactionId(Long toTransactionId) {
this.toTransactionId = toTransactionId;
}
public ReportRequest toTransactionId(Integer toTransactionId) {
public ReportRequest toTransactionId(Long toTransactionId) {
this.toTransactionId = toTransactionId;
return this;
}

View File

@@ -2,21 +2,17 @@ package com.inteligr8.alfresco.asie.model.core;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.annotation.JsonProperty.Access;
import com.inteligr8.alfresco.asie.model.BaseResponse;
import com.inteligr8.solr.model.Cores;
@JsonIgnoreProperties(ignoreUnknown = true)
public class ReportResponse extends BaseResponse {
@JsonProperty(access = Access.READ_ONLY)
private Report report;
@JsonProperty(required = true)
private Cores<Report> cores;
public Report getReport() {
return report;
}
protected void setReport(Report report) {
this.report = report;
public Cores<Report> getCores() {
return cores;
}
}

View File

@@ -0,0 +1,17 @@
package com.inteligr8.alfresco.asie.model.core;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.inteligr8.solr.model.Action;
@JsonIgnoreProperties(ignoreUnknown = true)
public class RetryAction extends Action {
@JsonProperty(value = "alfresco")
private int[] nodeIds;
public int[] getNodeIds() {
return nodeIds;
}
}

View File

@@ -1,22 +0,0 @@
package com.inteligr8.alfresco.asie.model.core;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.annotation.JsonProperty.Access;
import com.inteligr8.solr.model.ResponseAction;
@JsonIgnoreProperties(ignoreUnknown = true)
public class RetryResponseAction extends ResponseAction {
@JsonProperty(value = "alfresco", access = Access.READ_ONLY)
private int[] nodeIds;
public int[] getNodeIds() {
return nodeIds;
}
public void setNodeIds(int[] nodeIds) {
this.nodeIds = nodeIds;
}
}

View File

@@ -1,24 +1,7 @@
package com.inteligr8.alfresco.asie.model.core;
import java.util.Map;
import com.inteligr8.solr.model.Metadata;
import com.fasterxml.jackson.annotation.JsonAnyGetter;
import com.fasterxml.jackson.annotation.JsonAnySetter;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
@JsonIgnoreProperties(ignoreUnknown = true)
public class Summary {
private Map<String, Object> summary;
@JsonAnyGetter
public Map<String, Object> getSummary() {
return summary;
}
@JsonAnySetter
public void setSummary(Map<String, Object> summary) {
this.summary = summary;
}
public class Summary extends Metadata {
}

View File

@@ -2,21 +2,17 @@ package com.inteligr8.alfresco.asie.model.core;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.annotation.JsonProperty.Access;
import com.inteligr8.alfresco.asie.model.BaseResponse;
import com.inteligr8.solr.model.Cores;
@JsonIgnoreProperties(ignoreUnknown = true)
public class SummaryResponse extends BaseResponse {
@JsonProperty(value = "Summary", access = Access.READ_ONLY)
private Summary summary;
@JsonProperty(value = "Summary", required = true)
private Cores<Summary> cores;
public Summary getSummary() {
return summary;
}
public void setSummary(Summary summary) {
this.summary = summary;
public Cores<Summary> getCores() {
return cores;
}
}

View File

@@ -0,0 +1,24 @@
package com.inteligr8.alfresco.asie;
import java.net.URL;
import com.inteligr8.alfresco.asie.api.CoreAdminApi;
import com.inteligr8.rs.ClientJerseyImpl;
public class AsieClient extends ClientJerseyImpl {
public AsieClient(String hostname) {
super(new AsieClientConfiguration().withHostname(hostname));
this.register();
}
public AsieClient(URL baseUrl) {
super(new AsieClientConfiguration().withBaseUrl(baseUrl.toString()));
this.register();
}
public CoreAdminApi getCoreAdminApi() {
return this.getApi(this.getConfig().createAuthorizationFilter(), CoreAdminApi.class);
}
}

View File

@@ -0,0 +1,54 @@
package com.inteligr8.alfresco.asie;
import java.io.IOException;
import java.util.Arrays;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.inteligr8.rs.AuthorizationFilter;
import com.inteligr8.rs.ClientJerseyConfiguration;
import jakarta.ws.rs.client.ClientRequestContext;
public class AsieClientConfiguration implements ClientJerseyConfiguration {
private Logger logger = LoggerFactory.getLogger(AsieClientConfiguration.class);
private String baseUrl = "http://locahost:8983/solr";
private String searchSecret = "alfresco-secret";
@Override
public String getBaseUrl() {
return this.baseUrl;
}
public AsieClientConfiguration withBaseUrl(String baseUrl) {
this.baseUrl = baseUrl;
return this;
}
public AsieClientConfiguration withHostname(String hostname) {
this.baseUrl = "http://" + hostname + ":8983/solr";
return this;
}
public AsieClientConfiguration withSearchSecret(String searchSecret) {
this.searchSecret = searchSecret;
return this;
}
@Override
public AuthorizationFilter createAuthorizationFilter() {
if (this.searchSecret == null)
return null;
return new AuthorizationFilter() {
@Override
public void filter(ClientRequestContext requestContext) throws IOException {
logger.trace("Adding ASIE secret for authorization ...");
requestContext.getHeaders().addAll("X-Alfresco-Search-Secret", Arrays.asList(searchSecret));
}
};
}
}

View File

@@ -0,0 +1,27 @@
package com.inteligr8.alfresco.asie;
import org.junit.jupiter.api.Assertions;
public class AssertionUtil {
public static <T> T assertNotNull(T obj) {
Assertions.assertNotNull(obj);
return obj;
}
public static <T> T assertNotNull(T obj, String message) {
Assertions.assertNotNull(obj, message);
return obj;
}
public static void assertType(Object obj, Class<?> type) {
Assertions.assertNotNull(obj);
Assertions.assertEquals(type, obj.getClass());
}
public static void assertType(Object obj, Class<?> type, String message) {
Assertions.assertNotNull(obj, message);
Assertions.assertEquals(type, obj.getClass(), message);
}
}

View File

@@ -0,0 +1,23 @@
package com.inteligr8.alfresco.asie.api;
import java.net.MalformedURLException;
import org.junit.jupiter.api.BeforeAll;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.inteligr8.alfresco.asie.AsieClient;
public class AbstractApiUnitTest {
protected Logger logger = LoggerFactory.getLogger(this.getClass());
protected static AsieClient client;
protected static String defaultCore = "alfresco";
@BeforeAll
private static void init() throws MalformedURLException {
client = new AsieClient("localhost");
}
}

View File

@@ -0,0 +1,34 @@
package com.inteligr8.alfresco.asie.api;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import com.inteligr8.alfresco.asie.AssertionUtil;
import com.inteligr8.alfresco.asie.model.ActionCoreResponse;
import com.inteligr8.alfresco.asie.model.core.ReindexRequest;
import com.inteligr8.solr.model.Action;
import com.inteligr8.solr.model.Action.Status;
import com.inteligr8.solr.model.Cores;
import com.inteligr8.solr.model.ResponseHeader;
public class CoreAdminReindexUnitTest extends AbstractApiUnitTest {
@Test
public void reindex() {
CoreAdminApi api = client.getCoreAdminApi();
ActionCoreResponse<Action> response = api.reindex(
new ReindexRequest()
.withCore(defaultCore));
Assertions.assertNotNull(response);
ResponseHeader responseHeader = AssertionUtil.assertNotNull(response.getResponseHeader());
Assertions.assertEquals(0, responseHeader.getStatus());
Cores<Action> cores = AssertionUtil.assertNotNull(response.getCores());
Action action = AssertionUtil.assertNotNull(cores.getByCore(defaultCore));
Assertions.assertEquals(Status.Scheduled, action.getStatus());
}
}

View File

@@ -0,0 +1,96 @@
package com.inteligr8.alfresco.asie.api;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import com.inteligr8.alfresco.asie.AsieClient;
import com.inteligr8.alfresco.asie.AssertionUtil;
import com.inteligr8.alfresco.asie.model.core.Report;
import com.inteligr8.alfresco.asie.model.core.ReportRequest;
import com.inteligr8.alfresco.asie.model.core.ReportResponse;
import com.inteligr8.alfresco.asie.model.core.Summary;
import com.inteligr8.alfresco.asie.model.core.SummaryRequest;
import com.inteligr8.alfresco.asie.model.core.SummaryResponse;
import com.inteligr8.solr.model.CoreMetadata;
import com.inteligr8.solr.model.Cores;
import com.inteligr8.solr.model.ResponseHeader;
import com.inteligr8.solr.model.core.StatusRequest;
import com.inteligr8.solr.model.core.StatusResponse;
import jakarta.ws.rs.ProcessingException;
public class CoreAdminStatusUnitTest extends AbstractApiUnitTest {
@Test
public void noHost() {
AsieClient client = new AsieClient("doesnotexist");
CoreAdminApi api = client.getCoreAdminApi();
Assertions.assertThrows(ProcessingException.class, () -> {
api.getStatus(
new StatusRequest()
.withCore(defaultCore));
});
}
@Test
public void summary() {
CoreAdminApi api = client.getCoreAdminApi();
SummaryResponse response = api.getSummary(
new SummaryRequest()
.withCore(defaultCore));
Assertions.assertNotNull(response);
ResponseHeader responseHeader = AssertionUtil.assertNotNull(response.getResponseHeader());
Assertions.assertEquals(0, responseHeader.getStatus());
Cores<Summary> cores = AssertionUtil.assertNotNull(response.getCores());
Summary summary = AssertionUtil.assertNotNull(cores.getByCore(defaultCore));
AssertionUtil.assertType(summary.getByField("Active"), Boolean.class);
AssertionUtil.assertType(summary.getByField("Number of Searchers"), Integer.class);
AssertionUtil.assertType(summary.getByField("Last Index TX Commit Time"), Long.class);
AssertionUtil.assertType(summary.getByField("Last Index TX Commit Date"), String.class);
}
@Test
public void status() {
CoreAdminApi api = client.getCoreAdminApi();
StatusResponse response = api.getStatus(
new StatusRequest()
.withCore(defaultCore));
Assertions.assertNotNull(response);
ResponseHeader responseHeader = AssertionUtil.assertNotNull(response.getResponseHeader());
Assertions.assertEquals(0, responseHeader.getStatus());
Cores<CoreMetadata> cores = AssertionUtil.assertNotNull(response.getCores());
CoreMetadata core = AssertionUtil.assertNotNull(cores.getByCore(defaultCore));
Assertions.assertEquals(core, core.getName());
Assertions.assertNotNull(core.getStartDateTime());
}
@Test
public void report() {
CoreAdminApi api = client.getCoreAdminApi();
ReportResponse response = api.getReport(
new ReportRequest()
.withCore(defaultCore));
Assertions.assertNotNull(response);
ResponseHeader responseHeader = AssertionUtil.assertNotNull(response.getResponseHeader());
Assertions.assertEquals(0, responseHeader.getStatus());
Cores<Report> cores = AssertionUtil.assertNotNull(response.getCores());
Report report = AssertionUtil.assertNotNull(cores.getByCore(defaultCore));
AssertionUtil.assertType(report.getByField("Index error count"), Integer.class);
AssertionUtil.assertType(report.getByField("Last indexed transaction commit time"), Long.class);
AssertionUtil.assertType(report.getByField("Last indexed transaction commit date"), String.class);
}
}

View File

@@ -0,0 +1,23 @@
rootLogger.level=trace
rootLogger.appenderRef.stdout.ref=STDOUT
logger.inteligr8-rs-request.name=jaxrs.request
logger.inteligr8-rs-request.level=trace
logger.inteligr8-rs-response.name=jaxrs.response
logger.inteligr8-rs-response.level=off
logger.this.name=com.inteligr8.alfresco.asie
logger.this.level=trace
# hide framework
logger.apache-http.name=org.apache.http
logger.apache-http.level=debug
logger.jersey.name=org.glassfish.jersey
logger.jersey.level=trace
logger.jersey-client.name=org.glassfish.jersey.client
logger.jersey-client.level=trace
appender.stdout.type=Console
appender.stdout.name=STDOUT
appender.stdout.layout.type=PatternLayout
appender.stdout.layout.pattern=%d{ABSOLUTE_MICROS} %level{length=1} %c{1}: %m%n

View File

@@ -6,7 +6,7 @@
<parent>
<groupId>com.inteligr8.alfresco</groupId>
<artifactId>asie-platform-module-parent</artifactId>
<version>1.2.1</version>
<version>1.2.2</version>
<relativePath>../</relativePath>
</parent>

View File

@@ -6,7 +6,7 @@
<parent>
<groupId>com.inteligr8.alfresco</groupId>
<artifactId>asie-platform-module-parent</artifactId>
<version>1.2.1</version>
<version>1.2.2</version>
<relativePath>../</relativePath>
</parent>
@@ -83,7 +83,7 @@
<dependency>
<groupId>com.inteligr8.alfresco</groupId>
<artifactId>cxf-jaxrs-platform-module</artifactId>
<version>1.3.1-acs-v23.3</version>
<version>1.3.2-acs-v23.3</version>
<type>amp</type>
</dependency>

View File

@@ -91,7 +91,7 @@ public abstract class AbstractUnregisterNodeWebScript<T extends NodeParameterSet
if (status == null) {
this.logger.warn("Registered host/core status could not be retrieved: {}:{}/solr/{}", nodeHostname, nodePort, core);
} else {
CoreMetadata coreMetadata = status.getStatus().getCores().get(core);
CoreMetadata coreMetadata = status.getCores().getByCore(core);
if (coreMetadata == null || coreMetadata.getName() == null) {
this.logger.warn("Registered core does not actually exist on the node host; could be a DNS issue: {}:{}/solr/{}", nodeHostname, nodePort, core);
} else {
@@ -141,13 +141,13 @@ public abstract class AbstractUnregisterNodeWebScript<T extends NodeParameterSet
protected StatusResponse getCoreStatus(String nodeHostname, int nodePort, String core) {
this.logger.debug("Retrieving status for core {} on ASIE node: {}", core, nodeHostname);
CoreAdminApi api = this.createApi(nodeHostname, nodePort);
CoreAdminApi api = this.getApiService().createApi(nodeHostname, nodePort, CoreAdminApi.class);
return api.getStatus(new StatusRequest().withCore(core));
}
protected void unloadCore(String nodeHostname, int nodePort, String core) {
this.logger.info("Unloading core {} on ASIE node: {}", core, nodeHostname);
CoreAdminApi api = this.createApi(nodeHostname, nodePort);
CoreAdminApi api = this.getApiService().createApi(nodeHostname, nodePort, CoreAdminApi.class);
api.unload(new UnloadRequest().withCore(core));
}

View File

@@ -63,7 +63,7 @@ public class ReloadNodeShardWebScript extends AbstractAsieNodeWebScript {
throw new WebScriptException(HttpStatus.NOT_FOUND.value(), "The specified node/shard could not be found or formulated");
this.logger.info("Reloading core {} on ASIE node: {}", coreName, nodeHostname);
CoreAdminApi api = this.createApi(nodeHostname, nodePort);
CoreAdminApi api = this.getApiService().createApi(nodeHostname, nodePort, CoreAdminApi.class);
try {
api.create(new CreateRequest()
.withCore(coreName)

View File

@@ -64,7 +64,7 @@ public class ReloadNodeWebScript extends AbstractAsieNodeWebScript {
String coreInstancePath = core.getValue();
this.logger.info("Reloading core {} on ASIE node: {}", coreName, nodeHostname);
CoreAdminApi api = this.createApi(nodeHostname, nodePort);
CoreAdminApi api = this.getApiService().createApi(nodeHostname, nodePort, CoreAdminApi.class);
try {
api.create(new CreateRequest()
.withCore(coreName)

35
pom.xml
View File

@@ -5,7 +5,7 @@
<groupId>com.inteligr8.alfresco</groupId>
<artifactId>asie-platform-module-parent</artifactId>
<version>1.2.1</version>
<version>1.2.2</version>
<packaging>pom</packaging>
<name>ASIE Platform Module Parent</name>
@@ -39,9 +39,9 @@
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<maven.compiler.source>11</maven.compiler.source>
<maven.compiler.target>11</maven.compiler.target>
<maven.compiler.release>11</maven.compiler.release>
<maven.compiler.source>17</maven.compiler.source>
<maven.compiler.target>17</maven.compiler.target>
<maven.compiler.release>17</maven.compiler.release>
<maven.deploy.skip>true</maven.deploy.skip>
</properties>
@@ -56,12 +56,34 @@
<!-- avoids struts dependency -->
<plugin>
<artifactId>maven-site-plugin</artifactId>
<version>3.12.1</version>
<version>3.21.0</version>
</plugin>
<!-- Force use of a new maven-dependency-plugin that doesn't download struts dependency -->
<plugin>
<artifactId>maven-dependency-plugin</artifactId>
<version>3.8.0</version>
<version>3.8.1</version>
</plugin>
<plugin>
<artifactId>maven-surefire-plugin</artifactId>
<version>3.4.0</version>
<dependencies>
<dependency>
<groupId>org.junit.jupiter</groupId>
<artifactId>junit-jupiter-engine</artifactId>
<version>5.11.0</version>
</dependency>
</dependencies>
</plugin>
<plugin>
<artifactId>maven-failsafe-plugin</artifactId>
<version>3.4.0</version>
<dependencies>
<dependency>
<groupId>org.junit.jupiter</groupId>
<artifactId>junit-jupiter-engine</artifactId>
<version>5.11.0</version>
</dependency>
</dependencies>
</plugin>
</plugins>
</pluginManagement>
@@ -72,7 +94,6 @@
<module>asie-api</module>
<module>shared</module>
<module>enterprise-module</module>
<module>community-module</module>
</modules>
<profiles>

View File

@@ -6,7 +6,7 @@
<parent>
<groupId>com.inteligr8.alfresco</groupId>
<artifactId>asie-platform-module-parent</artifactId>
<version>1.2.1</version>
<version>1.2.2</version>
<relativePath>../</relativePath>
</parent>
@@ -41,10 +41,15 @@
<dependency>
<groupId>com.inteligr8</groupId>
<artifactId>common-rest-client</artifactId>
<version>3.0.1-cxf</version>
<version>3.0.3-cxf</version>
</dependency>
<!-- Needed by this module, but provided by ACS -->
<dependency>
<groupId>org.alfresco</groupId>
<artifactId>alfresco-data-model</artifactId>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.alfresco</groupId>
<artifactId>alfresco-repository</artifactId>

View File

@@ -0,0 +1,36 @@
package com.inteligr8.alfresco.asie.model;
import java.io.Serializable;
public class PersistedNode implements Serializable {
private static final long serialVersionUID = 4105196543023419818L;
private final SolrHost node;
private final long persistMillis;
private long expireTimeMillis;
public PersistedNode(SolrHost node, int persistMinutes) {
this.node = node;
this.persistMillis = persistMinutes * 60L * 1000L;
this.reset();
}
public void reset() {
this.expireTimeMillis = System.currentTimeMillis() + this.persistMillis;
}
public boolean isExpired() {
return this.expireTimeMillis < System.currentTimeMillis();
}
public SolrHost getNode() {
return this.node;
}
@Override
public String toString() {
return "node: " + this.node + "; expires in: " + (System.currentTimeMillis() - this.expireTimeMillis) + " ms";
}
}

View File

@@ -19,7 +19,7 @@ public class Shard implements Serializable {
private final String spec;
protected Shard(ShardSet shardSet, int shardId) {
this.spec = shardSet.getCore() + "~" + shardId;
this.spec = shardSet.getCore() + "-" + shardId;
}
protected Shard(String spec) {
@@ -34,16 +34,20 @@ public class Shard implements Serializable {
}
public String getSpec() {
return spec;
return this.spec;
}
public String getCoreName() {
return this.spec;
}
public String extractShardSetCore() {
int pos = this.spec.indexOf('~');
int pos = this.spec.lastIndexOf('-');
return this.spec.substring(0, pos);
}
public int extractShardId() {
int pos = this.spec.indexOf('~');
int pos = this.spec.lastIndexOf('-');
return Integer.parseInt(this.spec.substring(pos+1));
}

View File

@@ -13,7 +13,7 @@ public class ShardInstance implements Serializable {
private final String spec;
protected ShardInstance(Shard shard, SolrHost node) {
this.spec = shard.getSpec() + "~" + node.getSpec();
this.spec = node.getSpec() + "~" + shard.getSpec();
}
public org.alfresco.repo.index.shard.ShardInstance toAlfrescoModel(org.alfresco.repo.index.shard.Shard shard) {
@@ -33,14 +33,14 @@ public class ShardInstance implements Serializable {
return spec;
}
public Shard extractShard() {
int pos = this.spec.indexOf('~');
return Shard.from(this.spec.substring(0, pos));
}
public SolrHost extractNode() {
int pos = this.spec.indexOf('~');
return SolrHost.from(this.spec.substring(pos+1));
return SolrHost.from(this.spec.substring(0, pos));
}
public Shard extractShard() {
int pos = this.spec.indexOf('~');
return Shard.from(this.spec.substring(pos+1));
}
@Override

View File

@@ -41,6 +41,8 @@ public class SolrHost implements Serializable {
this.spec = spec;
Matcher matcher = PATTERN.matcher(spec);
if (!matcher.find())
throw new IllegalArgumentException();
this.hostname = matcher.group(1);
this.port = Integer.parseInt(matcher.group(2));
this.path = matcher.group(3);

View File

@@ -0,0 +1,119 @@
package com.inteligr8.alfresco.asie.rest;
import java.io.IOException;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.alfresco.model.ContentModel;
import org.alfresco.service.cmr.repository.InvalidNodeRefException;
import org.alfresco.service.cmr.repository.NodeRef;
import org.alfresco.service.cmr.repository.NodeService;
import org.alfresco.service.cmr.repository.StoreRef;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.extensions.webscripts.WebScriptException;
import org.springframework.extensions.webscripts.WebScriptRequest;
import org.springframework.extensions.webscripts.WebScriptResponse;
import org.springframework.http.HttpStatus;
import com.inteligr8.alfresco.asie.model.ShardInstance;
import com.inteligr8.alfresco.asie.spi.ActionCallback;
public abstract class AbstractAcsNodeActionWebScript extends AbstractAsieWebScript {
private final Logger logger = LoggerFactory.getLogger(this.getClass());
@Autowired
private NodeService nodeService;
@Override
public void executeAuthorized(WebScriptRequest request, WebScriptResponse response) throws IOException {
String nodeId = request.getServiceMatch().getTemplateVars().get("nodeId");
NodeRef nodeRef = new NodeRef(StoreRef.STORE_REF_WORKSPACE_SPACESSTORE, nodeId);
long nodeDbId = this.findNodeDbId(nodeRef);
this.logger.trace("Found node database ID: {}: {}", nodeId, nodeDbId);
try {
Map<String, Object> responseMap = new HashMap<>();
responseMap.put("nodeDbId", nodeDbId);
ActionCallback callback = new ActionCallback() {
@Override
public void success(ShardInstance instance) {
@SuppressWarnings("unchecked")
List<String> instances = (List<String>) responseMap.get("success");
if (instances == null)
responseMap.put("success", instances = new LinkedList<>());
instances.add(instance.getSpec());
}
@Override
public void scheduled(ShardInstance instance) {
@SuppressWarnings("unchecked")
List<String> instances = (List<String>) responseMap.get("scheduled");
if (instances == null)
responseMap.put("scheduled", instances = new LinkedList<>());
instances.add(instance.getSpec());
}
@Override
public void error(ShardInstance instance, String message) {
@SuppressWarnings("unchecked")
Map<String, Object> instances = (Map<String, Object>) responseMap.get("error");
if (instances == null)
responseMap.put("error", instances = new HashMap<>());
instances.put(instance.getSpec(), Collections.singletonMap("message", message));
}
@Override
public void unknownResult(ShardInstance instance) {
@SuppressWarnings("unchecked")
List<String> instances = (List<String>) responseMap.get("unknown");
if (instances == null)
responseMap.put("unknown", instances = new LinkedList<>());
instances.add(instance.getSpec());
}
};
this.executeAction(nodeDbId, callback, 10L, TimeUnit.SECONDS, 30L, TimeUnit.SECONDS);
if (responseMap.containsKey("error")) {
response.setStatus(HttpStatus.INTERNAL_SERVER_ERROR.value());
} else if (responseMap.containsKey("scheduled")) {
response.setStatus(HttpStatus.ACCEPTED.value());
} else {
response.setStatus(HttpStatus.OK.value());
}
response.setContentType("application/json");
this.getObjectMapper().writeValue(response.getWriter(), responseMap);
} catch (UnsupportedOperationException uoe) {
throw new WebScriptException(HttpStatus.NOT_IMPLEMENTED.value(), uoe.getMessage(), uoe);
} catch (InterruptedException ie) {
throw new WebScriptException(HttpStatus.SERVICE_UNAVAILABLE.value(), "The execution was interrupted", ie);
} catch (TimeoutException te) {
throw new WebScriptException(HttpStatus.INTERNAL_SERVER_ERROR.value(), "The execution may continue, but timed-out waiting", te);
}
}
protected abstract void executeAction(
long nodeDbId, ActionCallback callback,
long fullQueueTimeout, TimeUnit fullQueueUnit,
long execTimeout, TimeUnit execUnit) throws TimeoutException, InterruptedException;
private long findNodeDbId(NodeRef nodeRef) {
try {
return (Long) this.nodeService.getProperty(nodeRef, ContentModel.PROP_NODE_DBID);
} catch (InvalidNodeRefException inre) {
throw new WebScriptException(HttpStatus.NOT_FOUND.value(), "The node does not exist");
}
}
}

View File

@@ -0,0 +1,102 @@
package com.inteligr8.alfresco.asie.rest;
import java.io.IOException;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.alfresco.model.ContentModel;
import org.alfresco.service.cmr.repository.InvalidNodeRefException;
import org.alfresco.service.cmr.repository.NodeRef;
import org.alfresco.service.cmr.repository.NodeService;
import org.alfresco.service.cmr.repository.StoreRef;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.extensions.webscripts.WebScriptException;
import org.springframework.extensions.webscripts.WebScriptRequest;
import org.springframework.extensions.webscripts.WebScriptResponse;
import org.springframework.http.HttpStatus;
import com.inteligr8.alfresco.asie.model.ShardInstance;
import com.inteligr8.alfresco.asie.spi.ActionCallback;
public abstract class AbstractActionWebScript extends AbstractAsieWebScript {
private final Logger logger = LoggerFactory.getLogger(this.getClass());
@Override
public void executeAuthorized(WebScriptRequest request, WebScriptResponse response) throws IOException {
try {
Map<String, Object> responseMap = new HashMap<>();
ActionCallback callback = new ActionCallback() {
@Override
public void success(ShardInstance instance) {
@SuppressWarnings("unchecked")
List<String> instances = (List<String>) responseMap.get("success");
if (instances == null)
responseMap.put("success", instances = new LinkedList<>());
instances.add(instance.getSpec());
}
@Override
public void scheduled(ShardInstance instance) {
@SuppressWarnings("unchecked")
List<String> instances = (List<String>) responseMap.get("scheduled");
if (instances == null)
responseMap.put("scheduled", instances = new LinkedList<>());
instances.add(instance.getSpec());
}
@Override
public void error(ShardInstance instance, String message) {
@SuppressWarnings("unchecked")
Map<String, Object> instances = (Map<String, Object>) responseMap.get("error");
if (instances == null)
responseMap.put("error", instances = new HashMap<>());
instances.put(instance.getSpec(), Collections.singletonMap("message", message));
}
@Override
public void unknownResult(ShardInstance instance) {
@SuppressWarnings("unchecked")
List<String> instances = (List<String>) responseMap.get("unknown");
if (instances == null)
responseMap.put("unknown", instances = new LinkedList<>());
instances.add(instance.getSpec());
}
};
this.executeAction(callback, 10L, TimeUnit.SECONDS, 30L, TimeUnit.SECONDS);
if (responseMap.containsKey("error")) {
response.setStatus(HttpStatus.INTERNAL_SERVER_ERROR.value());
} else if (responseMap.containsKey("scheduled")) {
response.setStatus(HttpStatus.ACCEPTED.value());
} else {
response.setStatus(HttpStatus.OK.value());
}
response.setContentType("application/json");
this.getObjectMapper().writeValue(response.getWriter(), responseMap);
} catch (UnsupportedOperationException uoe) {
throw new WebScriptException(HttpStatus.NOT_IMPLEMENTED.value(), uoe.getMessage(), uoe);
} catch (InterruptedException ie) {
throw new WebScriptException(HttpStatus.SERVICE_UNAVAILABLE.value(), "The execution was interrupted", ie);
} catch (TimeoutException te) {
throw new WebScriptException(HttpStatus.INTERNAL_SERVER_ERROR.value(), "The execution may continue, but timed-out waiting", te);
}
}
protected abstract void executeAction(
ActionCallback callback,
long fullQueueTimeout, TimeUnit fullQueueUnit,
long execTimeout, TimeUnit execUnit) throws TimeoutException, InterruptedException;
}

View File

@@ -20,7 +20,7 @@ public abstract class AbstractAsieNodeShardWebScript extends AbstractAsieShardab
String nodeEndpoint = this.getRequiredPathParameter(req, "nodeEndpoint");
int colon = nodeEndpoint.lastIndexOf(':');
String nodeHostname = colon < 0 ? nodeEndpoint : nodeEndpoint.substring(0, colon);
int nodePort = colon < 0 ? this.getDefaultSolrPort() : Integer.parseInt(nodeEndpoint.substring(colon+1));
int nodePort = colon < 0 ? this.getApiService().getDefaultSolrPort() : Integer.parseInt(nodeEndpoint.substring(colon+1));
ShardSet shardSet = this.getRequiredPathParameter(req, "shardSet", ShardSet.class);
int shardId = this.getRequiredPathParameter(req, "shardId", Integer.class);

View File

@@ -31,7 +31,7 @@ public abstract class AbstractAsieNodeWebScript extends AbstractAsieShardableWeb
int colon = nodeEndpoint.lastIndexOf(':');
String nodeHostname = colon < 0 ? nodeEndpoint : nodeEndpoint.substring(0, colon);
nodeHostname = nodeHostname.replace('_', '.');
int nodePort = colon < 0 ? this.getDefaultSolrPort() : Integer.parseInt(nodeEndpoint.substring(colon+1));
int nodePort = colon < 0 ? this.getApiService().getDefaultSolrPort() : Integer.parseInt(nodeEndpoint.substring(colon+1));
this.execute(req, res, nodeHostname, nodePort);
}

View File

@@ -122,7 +122,7 @@ public abstract class AbstractAsieShardableWebScript extends AbstractAsieWebScri
}
protected CoreAdminApi getApi(ShardInstance shard) {
return this.createApi(shard.getHostName(), shard.getPort());
return this.getApiService().createApi(shard.getHostName(), shard.getPort(), CoreAdminApi.class);
}
}

View File

@@ -1,140 +1,34 @@
package com.inteligr8.alfresco.asie.rest;
import java.io.IOException;
import java.util.HashSet;
import java.util.Set;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.extensions.webscripts.WebScriptRequest;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.inteligr8.alfresco.asie.Constants;
import com.inteligr8.alfresco.asie.api.CoreAdminApi;
import com.inteligr8.rs.AuthorizationFilter;
import com.inteligr8.rs.Client;
import com.inteligr8.rs.ClientCxfConfiguration;
import com.inteligr8.rs.ClientCxfImpl;
import com.inteligr8.alfresco.asie.service.ApiService;
import jakarta.ws.rs.client.ClientRequestContext;
public abstract class AbstractAsieWebScript extends AbstractWebScript implements InitializingBean {
private final Logger logger = LoggerFactory.getLogger(this.getClass());
@Value("${solr.secureComms}")
private String solrSecureComms;
@Value("${solr.port}")
private int solrPort;
@Value("${solr.port.ssl}")
private int solrSslPort;
@Value("${solr.sharedSecret.header}")
private String solrSharedSecretHeader;
@Value("${solr.sharedSecret}")
private String solrSharedSecret;
@Value("${inteligr8.asie.allowedAuthorities}")
private String authorizedAuthoritiesStr;
@Value("${inteligr8.asie.basePath}")
private String solrBaseUrl;
public abstract class AbstractAsieWebScript extends AbstractWebScript {
@Autowired
@Qualifier(Constants.QUALIFIER_ASIE)
private ObjectMapper objectMapper;
private Set<String> authorizedAuthorities;
@Override
public void afterPropertiesSet() throws Exception {
this.authorizedAuthorities = new HashSet<>();
String[] authorities = this.authorizedAuthoritiesStr.split(",");
for (String authority : authorities) {
authority = StringUtils.trimToNull(authority);
if (authority != null)
this.authorizedAuthorities.add(authority);
}
if (this.authorizedAuthorities.isEmpty())
this.logger.warn("All authenticated users will be authorized to access ASIE web scripts");
this.solrSharedSecret = StringUtils.trimToNull(this.solrSharedSecret);
}
@Override
protected Set<String> getAuthorities() {
return this.authorizedAuthorities;
}
@Autowired
private ApiService api;
protected ObjectMapper getObjectMapper() {
return this.objectMapper;
}
protected CoreAdminApi createApi(String hostname, int port) {
String solrBaseUrl = this.formulateSolrBaseUrl(hostname, port);
this.logger.trace("Using Solr base URL: {}", solrBaseUrl);
Client solrClient = this.createClient(solrBaseUrl);
return this.getApi(solrClient);
}
protected CoreAdminApi getApi(Client solrClient) {
return solrClient.getApi(CoreAdminApi.class);
}
protected int getDefaultSolrPort() {
boolean isSsl = "https".equals(this.solrSecureComms);
return isSsl ? this.solrSslPort : this.solrPort;
protected ApiService getApiService() {
return this.api;
}
protected String formulateSolrBaseUrl(WebScriptRequest req) {
String hostname = this.getRequiredPathParameter(req, "hostname");
Integer port = this.getOptionalPathParameter(req, "port", Integer.class);
return this.formulateSolrBaseUrl(hostname, port);
}
protected String formulateSolrBaseUrl(String hostname, Integer port) {
boolean isSsl = "https".equals(this.solrSecureComms);
StringBuilder baseUrl = new StringBuilder(isSsl ? "https" : "http").append("://").append(hostname);
baseUrl.append(':').append(port == null ? (isSsl ? this.solrSslPort : this.solrPort) : port);
baseUrl.append(this.solrBaseUrl);
return baseUrl.toString();
}
protected Client createClient(final String baseUrl) {
ClientCxfImpl client = new ClientCxfImpl(new ClientCxfConfiguration() {
@Override
public String getBaseUrl() {
return baseUrl.toString();
}
@Override
public AuthorizationFilter createAuthorizationFilter() {
return solrSharedSecret == null ? null : new AuthorizationFilter() {
@Override
public void filter(ClientRequestContext requestContext) throws IOException {
logger.debug("Adding authorization headers for ASIE shared auth: {}", solrSharedSecretHeader);
requestContext.getHeaders().putSingle(solrSharedSecretHeader, solrSharedSecret);
}
};
}
@Override
public boolean isDefaultBusEnabled() {
return false;
}
});
client.register();
return client;
return this.api.formulateSolrBaseUrl(hostname, port);
}
}

View File

@@ -4,11 +4,19 @@ import java.io.IOException;
import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.Collections;
import java.util.HashSet;
import java.util.Set;
import org.alfresco.repo.security.authentication.AuthenticationUtil;
import org.alfresco.service.cmr.security.AuthorityService;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.extensions.webscripts.Description.RequiredAuthentication;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.extensions.webscripts.WebScriptException;
import org.springframework.extensions.webscripts.WebScriptRequest;
import org.springframework.extensions.webscripts.WebScriptResponse;
@@ -16,9 +24,38 @@ import org.springframework.http.HttpStatus;
import net.sf.acegisecurity.GrantedAuthority;
public abstract class AbstractWebScript extends org.springframework.extensions.webscripts.AbstractWebScript {
public abstract class AbstractWebScript extends org.springframework.extensions.webscripts.AbstractWebScript implements InitializingBean {
protected abstract Set<String> getAuthorities();
private final Logger logger = LoggerFactory.getLogger(this.getClass());
@Value("${inteligr8.asie.allowedAuthorities}")
private String authorizedAuthoritiesStr;
@Autowired
private AuthorityService authorityService;
private Set<String> authorizedAuthorities;
@Override
public void afterPropertiesSet() throws Exception {
this.authorizedAuthorities = new HashSet<>();
String[] authorities = this.authorizedAuthoritiesStr.split(",");
for (String authority : authorities) {
authority = StringUtils.trimToNull(authority);
if (authority != null)
this.authorizedAuthorities.add(authority);
}
if (this.authorizedAuthorities.isEmpty()) {
this.logger.warn("All authenticated users will be authorized to access web scripts");
} else {
this.logger.debug("Allowing only authorities: {}", this.authorizedAuthorities);
}
}
protected Set<String> getAuthorities() {
return this.authorizedAuthorities;
}
@Override
public final void execute(WebScriptRequest request, WebScriptResponse response) throws IOException {
@@ -38,6 +75,13 @@ public abstract class AbstractWebScript extends org.springframework.extensions.w
return true;
}
Set<String> authorities = this.authorityService.getAuthoritiesForUser(AuthenticationUtil.getFullyAuthenticatedUser());
if (authorities != null) {
if (!Collections.disjoint(this.getAuthorities(), authorities))
return true;
}
this.logger.trace("Not authorized: user '{}'; authorities: {} + {}", AuthenticationUtil.getFullyAuthenticatedUser(), AuthenticationUtil.getFullAuthentication().getAuthorities(), authorities);
return false;
}

View File

@@ -3,7 +3,6 @@ package com.inteligr8.alfresco.asie.rest;
import java.io.IOException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.extensions.webscripts.AbstractWebScript;
import org.springframework.extensions.webscripts.WebScriptRequest;
import org.springframework.extensions.webscripts.WebScriptResponse;
import org.springframework.http.HttpStatus;
@@ -20,13 +19,13 @@ public class ClearRegistryWebScript extends AbstractWebScript {
@Autowired
private ShardStateService sss;
@Override
public void execute(WebScriptRequest req, WebScriptResponse res) throws IOException {
@Override
public void executeAuthorized(WebScriptRequest request, WebScriptResponse response) throws IOException {
this.sss.clear();
this.sbs.forget();
res.setStatus(HttpStatus.OK.value());
response.setStatus(HttpStatus.OK.value());
}
}

View File

@@ -0,0 +1,24 @@
package com.inteligr8.alfresco.asie.rest;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import com.inteligr8.alfresco.asie.service.FixService;
import com.inteligr8.alfresco.asie.spi.ActionCallback;
@Component(value = "webscript.com.inteligr8.alfresco.asie.fix.post")
public class FixWebScript extends AbstractActionWebScript {
@Autowired
private FixService fixSerivce;
@Override
protected void executeAction(ActionCallback callback, long fullQueueTimeout, TimeUnit fullQueueUnit,
long execTimeout, TimeUnit execUnit) throws TimeoutException, InterruptedException {
this.fixSerivce.fix(callback, 10L, TimeUnit.SECONDS, 30L, TimeUnit.SECONDS);
}
}

View File

@@ -0,0 +1,24 @@
package com.inteligr8.alfresco.asie.rest;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import com.inteligr8.alfresco.asie.service.PurgeService;
import com.inteligr8.alfresco.asie.spi.ActionCallback;
@Component(value = "webscript.com.inteligr8.alfresco.asie.purgeAcsNode.put")
public class PurgeAcsNodeWebScript extends AbstractAcsNodeActionWebScript {
@Autowired
private PurgeService purgeSerivce;
@Override
protected void executeAction(long nodeDbId, ActionCallback callback, long fullQueueTimeout, TimeUnit fullQueueUnit,
long execTimeout, TimeUnit execUnit) throws TimeoutException, InterruptedException {
this.purgeSerivce.purge(nodeDbId, callback, 10L, TimeUnit.SECONDS, 30L, TimeUnit.SECONDS);
}
}

View File

@@ -0,0 +1,128 @@
package com.inteligr8.alfresco.asie.rest;
import java.io.IOException;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.extensions.webscripts.WebScriptException;
import org.springframework.extensions.webscripts.WebScriptRequest;
import org.springframework.extensions.webscripts.WebScriptResponse;
import org.springframework.http.HttpStatus;
import org.springframework.stereotype.Component;
import com.inteligr8.alfresco.asie.model.ShardInstance;
import com.inteligr8.alfresco.asie.service.AcsReconcileService;
import com.inteligr8.alfresco.asie.spi.ReconcileCallback;
@Component(value = "webscript.com.inteligr8.alfresco.asie.reconcileAcsNodes.post")
public class ReconcileAcsNodesWebScript extends AbstractAsieWebScript {
@Autowired
private AcsReconcileService reconcileService;
@Override
public void executeAuthorized(WebScriptRequest request, WebScriptResponse response) throws IOException {
final int fromDbId = this.getRequestTemplateIntegerVariable(request, "fromDbId");
final int toDbId = this.getRequestTemplateIntegerVariable(request, "toDbId");
final boolean reindex = Boolean.TRUE.equals(this.getOptionalQueryParameter(request, "reindex", Boolean.class));
final boolean includeReconciled = Boolean.TRUE.equals(this.getOptionalQueryParameter(request, "includeReconciled", Boolean.class));
final Map<String, Object> responseMap = new HashMap<>();
ReconcileCallback callback = new ReconcileCallback() {
@Override
public void reconciled(long nodeDbId) {
if (includeReconciled) {
@SuppressWarnings("unchecked")
List<Long> unreconciledNodeDbIds = (List<Long>) responseMap.get("reconciled");
if (unreconciledNodeDbIds == null)
responseMap.put("reconciled", unreconciledNodeDbIds = new LinkedList<>());
unreconciledNodeDbIds.add(nodeDbId);
}
}
@Override
public void unreconciled(long nodeDbId) {
@SuppressWarnings("unchecked")
List<Long> unreconciledNodeDbIds = (List<Long>) responseMap.get("unreconciled");
if (unreconciledNodeDbIds == null)
responseMap.put("unreconciled", unreconciledNodeDbIds = new LinkedList<>());
unreconciledNodeDbIds.add(nodeDbId);
}
@Override
public void processed(long nodeDbId, Set<ShardInstance> instsReconciled, Set<ShardInstance> instsReconciling,
Map<ShardInstance, String> instsErrorMessages) {
if (!instsReconciled.isEmpty()) {
@SuppressWarnings("unchecked")
Map<Long, List<String>> nodeHosts = (Map<Long, List<String>>) responseMap.get("success");
if (nodeHosts == null)
responseMap.put("success", nodeHosts = new HashMap<>());
List<String> instances = new LinkedList<>();
for (ShardInstance instance : instsReconciled)
instances.add(instance.getSpec());
nodeHosts.put(nodeDbId, instances);
}
if (!instsReconciling.isEmpty()) {
@SuppressWarnings("unchecked")
Map<Long, List<String>> nodeHosts = (Map<Long, List<String>>) responseMap.get("scheduled");
if (nodeHosts == null)
responseMap.put("scheduled", nodeHosts = new HashMap<>());
List<String> instances = new LinkedList<>();
for (ShardInstance instance : instsReconciled)
instances.add(instance.getSpec());
nodeHosts.put(nodeDbId, instances);
}
if (!instsErrorMessages.isEmpty()) {
@SuppressWarnings("unchecked")
Map<Long, Map<String, Map<String, String>>> nodeHosts = (Map<Long, Map<String, Map<String, String>>>) responseMap.get("error");
if (nodeHosts == null)
responseMap.put("error", nodeHosts = new HashMap<>());
Map<String, Map<String, String>> nodeHost = new HashMap<>();
for (Entry<ShardInstance, String> message : instsErrorMessages.entrySet())
nodeHost.put(message.getKey().getSpec(), Collections.singletonMap("message", message.getValue()));
nodeHosts.put(nodeDbId, nodeHost);
}
}
};
try {
this.reconcileService.reconcile(fromDbId, toDbId, null, reindex, callback, 1L, TimeUnit.HOURS, 2L, TimeUnit.MINUTES);
if (responseMap.containsKey("error")) {
response.setStatus(HttpStatus.INTERNAL_SERVER_ERROR.value());
} else if (responseMap.containsKey("scheduled")) {
response.setStatus(HttpStatus.ACCEPTED.value());
} else {
response.setStatus(HttpStatus.OK.value());
}
response.setContentType("application/json");
this.getObjectMapper().writeValue(response.getWriter(), responseMap);
} catch (InterruptedException ie) {
throw new WebScriptException(HttpStatus.SERVICE_UNAVAILABLE.value(), "The reindex was interrupted", ie);
} catch (TimeoutException te) {
throw new WebScriptException(HttpStatus.INTERNAL_SERVER_ERROR.value(), "The reindex may continue, but timed-out waiting", te);
}
}
private int getRequestTemplateIntegerVariable(WebScriptRequest request, String templateVariableName) {
String str = request.getServiceMatch().getTemplateVars().get(templateVariableName);
return Integer.valueOf(str);
}
}

View File

@@ -0,0 +1,24 @@
package com.inteligr8.alfresco.asie.rest;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import com.inteligr8.alfresco.asie.service.ReindexService;
import com.inteligr8.alfresco.asie.spi.ActionCallback;
@Component(value = "webscript.com.inteligr8.alfresco.asie.reindexAcsNode.put")
public class ReindexAcsNodeWebScript extends AbstractAcsNodeActionWebScript {
@Autowired
private ReindexService reindexSerivce;
@Override
protected void executeAction(long nodeDbId, ActionCallback callback, long fullQueueTimeout, TimeUnit fullQueueUnit,
long execTimeout, TimeUnit execUnit) throws TimeoutException, InterruptedException {
this.reindexSerivce.reindex(nodeDbId, callback, 10L, TimeUnit.SECONDS, 30L, TimeUnit.SECONDS);
}
}

View File

@@ -0,0 +1,24 @@
package com.inteligr8.alfresco.asie.rest;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import com.inteligr8.alfresco.asie.service.RetryService;
import com.inteligr8.alfresco.asie.spi.ActionCallback;
@Component(value = "webscript.com.inteligr8.alfresco.asie.retry.post")
public class RetryWebScript extends AbstractActionWebScript {
@Autowired
private RetryService retrySerivce;
@Override
protected void executeAction(ActionCallback callback, long fullQueueTimeout, TimeUnit fullQueueUnit,
long execTimeout, TimeUnit execUnit) throws TimeoutException, InterruptedException {
this.retrySerivce.retry(callback, 10L, TimeUnit.SECONDS, 30L, TimeUnit.SECONDS);
}
}

View File

@@ -0,0 +1,236 @@
package com.inteligr8.alfresco.asie.service;
import java.net.URL;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.alfresco.model.ContentModel;
import org.alfresco.repo.index.shard.Floc;
import org.alfresco.repo.index.shard.Shard;
import org.alfresco.repo.index.shard.ShardInstance;
import org.alfresco.repo.index.shard.ShardRegistry;
import org.alfresco.repo.index.shard.ShardState;
import org.alfresco.service.cmr.repository.StoreRef;
import org.alfresco.service.cmr.search.SearchParameters;
import org.alfresco.service.cmr.search.SearchService;
import org.alfresco.service.namespace.NamespaceService;
import org.alfresco.service.namespace.QName;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import com.inteligr8.alfresco.asie.Constants;
import com.inteligr8.alfresco.asie.api.CoreAdminApi;
import com.inteligr8.alfresco.asie.model.ActionCoreResponse;
import com.inteligr8.alfresco.asie.model.ShardSet;
import com.inteligr8.alfresco.asie.model.SolrHost;
import com.inteligr8.alfresco.asie.spi.ActionCallback;
import com.inteligr8.alfresco.asie.util.CompositeFuture;
import com.inteligr8.alfresco.asie.util.ThrottledThreadPoolExecutor;
import com.inteligr8.solr.model.Action;
import com.inteligr8.solr.model.ActionResponse;
import com.inteligr8.solr.model.BaseResponse;
public abstract class AbstractActionService {
private final Logger logger = LoggerFactory.getLogger(this.getClass());
@Autowired
private NamespaceService namespaceService;
@Autowired
private ApiService apiService;
@Autowired
private ExecutorManager executorManager;
@Autowired(required = false)
@Qualifier(Constants.QUALIFIER_ASIE)
private ShardRegistry shardRegistry;
@Value("${inteligr8.asie.default.concurrentQueueSize:64}")
private int concurrentQueueSize;
@Value("${inteligr8.asie.default.concurrency:16}")
private int concurrency;
protected int getConcurrency() {
return this.concurrency;
}
protected int getConcurrentQueueSize() {
return this.concurrentQueueSize;
}
protected abstract String getThreadNamePrefix();
protected abstract String getActionName();
/**
* This method executes an action on the specified node in Solr using its
* ACS unique database identifier. The callback handles all the return
* values. This is the synchronous alternative to the other `action`
* method.
*
* There are two sets of parameters regarding timeouts. The queue timeouts
* are for how long the requesting thread should wait for a full queue to
* open up space for new executions. The execution timeouts are for how
* long the execution should be allowed to take once dequeued. There is no
* timeout for how long the execution is queued.
*
* @param callback A callback to process multiple returned values from the action.
* @param fullQueueTimeout A timeout for how long the calling thread should wait for space on the queue.
* @param fullQueueUnit The time units for the `queueTimeout`.
* @param execTimeout A timeout for the elapsed time the execution should take when dequeued.
* @param execUnit The time units for the `execTimeout`.
* @throws TimeoutException Either the queue or execution timeout lapsed.
* @throws InterruptedException The execution was interrupted (server shutdown).
*/
protected void action(ActionCallback callback, long fullQueueTimeout, TimeUnit fullQueueUnit, long execTimeout, TimeUnit execUnit) throws TimeoutException, InterruptedException {
long fullQueueExpireTimeMillis = System.currentTimeMillis() + fullQueueUnit.toMillis(fullQueueTimeout);
Future<Void> future = this._action(callback, fullQueueExpireTimeMillis);
try {
future.get(execTimeout, execUnit);
} catch (ExecutionException ee) {
this.logger.error("Reindex thread failed: " + ee.getMessage(), ee);
}
}
/**
* This method executes an action on the specified node in Solr using its
* ACS unique database identifier. The callback handles all the return
* values. This is the asynchronous alternative to the other `action`
* method.
*
* This method may block indefinitely when the queue is full. Once all
* executions are queued, this will return a single future representing all
* the executions.
*
* @param callback A callback to process multiple returned values from the execution.
* @return A reference to the future or active executing task.
* @throws InterruptedException The execution was interrupted (server shutdown).
*/
protected Future<Void> action(ActionCallback callback) throws InterruptedException {
try {
return this._action(callback, null);
} catch (TimeoutException te) {
throw new RuntimeException("This should never happen: " + te.getMessage(), te);
}
}
private Future<Void> _action(ActionCallback callback, Long fullQueueExpireTimeMillis) throws TimeoutException, InterruptedException {
List<com.inteligr8.alfresco.asie.model.ShardInstance> eligibleInstances = this.findPossibleShardInstances();
this.logger.debug("Will attempt to {} {} shard instances", this.getActionName(), eligibleInstances.size());
CompositeFuture<Void> future = new CompositeFuture<>();
ThrottledThreadPoolExecutor executor = this.executorManager.createThrottled(
this.getThreadNamePrefix(),
this.getConcurrency(), this.getConcurrency(), this.getConcurrentQueueSize(),
1L, TimeUnit.MINUTES);
for (final com.inteligr8.alfresco.asie.model.ShardInstance instance : eligibleInstances) {
this.logger.trace("Will attempt to {} shard instance: {}", this.getActionName(), instance);
Callable<Void> callable = new Callable<>() {
@Override
public Void call() {
String core = instance.extractShard().getCoreName();
SolrHost host = instance.extractNode();
URL url = host.toUrl(apiService.isSecure() ? "https" : "http");
CoreAdminApi api = apiService.createApi(url.toString(), CoreAdminApi.class);
try {
logger.debug("Performing {} of shard instance: {}", getActionName(), instance);
BaseResponse apiResponse = execute(api, core);
logger.trace("Performed {} of shard instance: {}", getActionName(), instance);
Action action = null;
if (apiResponse instanceof ActionCoreResponse<?>) {
action = ((ActionCoreResponse<Action>) apiResponse).getCores().getByCore(core);
} else if (apiResponse instanceof ActionResponse<?>) {
action = ((ActionResponse<Action>) apiResponse).getAction();
}
if (action == null) {
callback.unknownResult(instance);
} else {
switch (action.getStatus()) {
case Scheduled:
callback.scheduled(instance);
break;
case Success:
callback.success(instance);
break;
default:
if (apiResponse instanceof com.inteligr8.alfresco.asie.model.BaseResponse) {
com.inteligr8.alfresco.asie.model.BaseResponse asieResponse = (com.inteligr8.alfresco.asie.model.BaseResponse) apiResponse;
logger.debug("Performance of {} of shard instance failed: {}: {}", getActionName(), instance, asieResponse.getException());
callback.error(instance, asieResponse.getException());
} else {
logger.debug("Performance of {} of shard instance failed: {}: {}", getActionName(), instance, apiResponse.getResponseHeader().getStatus());
callback.error(instance, String.valueOf(apiResponse.getResponseHeader().getStatus()));
}
}
}
} catch (Exception e) {
logger.error("An exception occurred", e);
callback.error(instance, e.getMessage());
}
return null;
}
};
if (fullQueueExpireTimeMillis == null) {
future.combine(executor.submit(callable, -1L, null));
} else {
future.combine(executor.submit(callable, fullQueueExpireTimeMillis - System.currentTimeMillis(), TimeUnit.MILLISECONDS));
}
}
return future;
}
protected abstract BaseResponse execute(CoreAdminApi api, String core);
private List<com.inteligr8.alfresco.asie.model.ShardInstance> findPossibleShardInstances() {
if (this.shardRegistry == null)
throw new UnsupportedOperationException("ACS instances without a sharding configuration are not yet implemented");
List<com.inteligr8.alfresco.asie.model.ShardInstance> instances = new LinkedList<>();
for (Entry<Floc, Map<Shard, Set<ShardState>>> floc : this.shardRegistry.getFlocs().entrySet()) {
if (!floc.getKey().getStoreRefs().contains(StoreRef.STORE_REF_WORKSPACE_SPACESSTORE))
continue;
for (Entry<Shard, Set<ShardState>> shard : floc.getValue().entrySet()) {
for (ShardState shardState : shard.getValue())
instances.add(this.toModel(shardState.getShardInstance(), shardState));
}
}
this.logger.trace("Despite sharding, considering all shards and nodes: {}", instances);
return instances;
}
private com.inteligr8.alfresco.asie.model.ShardInstance toModel(ShardInstance instance, ShardState anyShardState) {
Floc floc = instance.getShard().getFloc();
ShardSet shardSet = ShardSet.from(floc, anyShardState);
SolrHost host = SolrHost.from(instance);
com.inteligr8.alfresco.asie.model.Shard shard = com.inteligr8.alfresco.asie.model.Shard.from(shardSet, instance.getShard().getInstance());
return com.inteligr8.alfresco.asie.model.ShardInstance.from(shard, host);
}
}

View File

@@ -0,0 +1,268 @@
package com.inteligr8.alfresco.asie.service;
import java.net.URL;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.alfresco.model.ContentModel;
import org.alfresco.repo.index.shard.Floc;
import org.alfresco.repo.index.shard.Shard;
import org.alfresco.repo.index.shard.ShardInstance;
import org.alfresco.repo.index.shard.ShardRegistry;
import org.alfresco.repo.index.shard.ShardState;
import org.alfresco.service.cmr.repository.StoreRef;
import org.alfresco.service.cmr.search.SearchParameters;
import org.alfresco.service.cmr.search.SearchService;
import org.alfresco.service.namespace.NamespaceService;
import org.alfresco.service.namespace.QName;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import com.inteligr8.alfresco.asie.Constants;
import com.inteligr8.alfresco.asie.api.CoreAdminApi;
import com.inteligr8.alfresco.asie.model.ActionCoreResponse;
import com.inteligr8.alfresco.asie.model.ShardSet;
import com.inteligr8.alfresco.asie.model.SolrHost;
import com.inteligr8.alfresco.asie.spi.ActionCallback;
import com.inteligr8.alfresco.asie.util.CompositeFuture;
import com.inteligr8.alfresco.asie.util.ThrottledThreadPoolExecutor;
import com.inteligr8.solr.model.Action;
import com.inteligr8.solr.model.ActionResponse;
import com.inteligr8.solr.model.BaseResponse;
public abstract class AbstractNodeActionService {
private final Logger logger = LoggerFactory.getLogger(this.getClass());
@Autowired
private NamespaceService namespaceService;
@Autowired
private ApiService apiService;
@Autowired
private ExecutorManager executorManager;
@Autowired(required = false)
@Qualifier(Constants.QUALIFIER_ASIE)
private ShardRegistry shardRegistry;
@Value("${inteligr8.asie.default.concurrentQueueSize:64}")
private int concurrentQueueSize;
@Value("${inteligr8.asie.default.concurrency:16}")
private int concurrency;
protected int getConcurrency() {
return this.concurrency;
}
protected int getConcurrentQueueSize() {
return this.concurrentQueueSize;
}
protected abstract String getThreadNamePrefix();
protected abstract String getActionName();
/**
* This method executes an action on the specified node in Solr using its
* ACS unique database identifier. The callback handles all the return
* values. This is the synchronous alternative to the other `action`
* method.
*
* There are two sets of parameters regarding timeouts. The queue timeouts
* are for how long the requesting thread should wait for a full queue to
* open up space for new executions. The execution timeouts are for how
* long the execution should be allowed to take once dequeued. There is no
* timeout for how long the execution is queued.
*
* @param nodeDbId A node database ID.
* @param callback A callback to process multiple returned values from the action.
* @param fullQueueTimeout A timeout for how long the calling thread should wait for space on the queue.
* @param fullQueueUnit The time units for the `queueTimeout`.
* @param execTimeout A timeout for the elapsed time the execution should take when dequeued.
* @param execUnit The time units for the `execTimeout`.
* @throws TimeoutException Either the queue or execution timeout lapsed.
* @throws InterruptedException The execution was interrupted (server shutdown).
*/
protected void action(long nodeDbId, ActionCallback callback, long fullQueueTimeout, TimeUnit fullQueueUnit, long execTimeout, TimeUnit execUnit) throws TimeoutException, InterruptedException {
long fullQueueExpireTimeMillis = System.currentTimeMillis() + fullQueueUnit.toMillis(fullQueueTimeout);
Future<Void> future = this._action(nodeDbId, callback, fullQueueExpireTimeMillis);
try {
future.get(execTimeout, execUnit);
} catch (ExecutionException ee) {
this.logger.error("Reindex thread failed: " + ee.getMessage(), ee);
}
}
/**
* This method executes an action on the specified node in Solr using its
* ACS unique database identifier. The callback handles all the return
* values. This is the asynchronous alternative to the other `action`
* method.
*
* This method may block indefinitely when the queue is full. Once all
* executions are queued, this will return a single future representing all
* the executions.
*
* @param nodeDbId A node database ID.
* @param callback A callback to process multiple returned values from the execution.
* @return A reference to the future or active executing task.
* @throws InterruptedException The execution was interrupted (server shutdown).
*/
protected Future<Void> action(long nodeDbId, ActionCallback callback) throws InterruptedException {
try {
return this._action(nodeDbId, callback, null);
} catch (TimeoutException te) {
throw new RuntimeException("This should never happen: " + te.getMessage(), te);
}
}
private Future<Void> _action(long nodeDbId, ActionCallback callback, Long fullQueueExpireTimeMillis) throws TimeoutException, InterruptedException {
List<com.inteligr8.alfresco.asie.model.ShardInstance> eligibleInstances = this.findPossibleShardInstances(nodeDbId);
this.logger.debug("Will attempt to {} ACS node against {} shard instances: {}", this.getActionName(), eligibleInstances.size(), nodeDbId);
CompositeFuture<Void> future = new CompositeFuture<>();
ThrottledThreadPoolExecutor executor = this.executorManager.createThrottled(
this.getThreadNamePrefix(),
this.getConcurrency(), this.getConcurrency(), this.getConcurrentQueueSize(),
1L, TimeUnit.MINUTES);
for (final com.inteligr8.alfresco.asie.model.ShardInstance instance : eligibleInstances) {
this.logger.trace("Will attempt to {} ACS node against shard instance: {}: {}", this.getActionName(), nodeDbId, instance);
Callable<Void> callable = new Callable<>() {
@Override
public Void call() {
String core = instance.extractShard().getCoreName();
SolrHost host = instance.extractNode();
URL url = host.toUrl(apiService.isSecure() ? "https" : "http");
CoreAdminApi api = apiService.createApi(url.toString(), CoreAdminApi.class);
try {
logger.debug("Performing {} of ACS node against shard instance: {}: {}", getActionName(), nodeDbId, instance);
BaseResponse apiResponse = execute(api, core, nodeDbId);
logger.trace("Performed {} of ACS node against shard instance: {}: {}", getActionName(), nodeDbId, instance);
Action action = null;
if (apiResponse instanceof ActionCoreResponse<?>) {
action = ((ActionCoreResponse<Action>) apiResponse).getCores().getByCore(core);
} else if (apiResponse instanceof ActionResponse<?>) {
action = ((ActionResponse<Action>) apiResponse).getAction();
}
if (action == null) {
callback.unknownResult(instance);
} else {
switch (action.getStatus()) {
case Scheduled:
callback.scheduled(instance);
break;
case Success:
callback.success(instance);
break;
default:
if (apiResponse instanceof com.inteligr8.alfresco.asie.model.BaseResponse) {
com.inteligr8.alfresco.asie.model.BaseResponse asieResponse = (com.inteligr8.alfresco.asie.model.BaseResponse) apiResponse;
logger.debug("Performance of {} of ACS node against shard instance failed: {}: {}: {}", getActionName(), nodeDbId, instance, asieResponse.getException());
callback.error(instance, asieResponse.getException());
} else {
logger.debug("Performance of {} of ACS node against shard instance failed: {}: {}: {}", getActionName(), nodeDbId, instance, apiResponse.getResponseHeader().getStatus());
callback.error(instance, String.valueOf(apiResponse.getResponseHeader().getStatus()));
}
}
}
} catch (Exception e) {
logger.error("An exception occurred", e);
callback.error(instance, e.getMessage());
}
return null;
}
};
if (fullQueueExpireTimeMillis == null) {
future.combine(executor.submit(callable, -1L, null));
} else {
future.combine(executor.submit(callable, fullQueueExpireTimeMillis - System.currentTimeMillis(), TimeUnit.MILLISECONDS));
}
}
return future;
}
protected abstract BaseResponse execute(CoreAdminApi api, String core, long nodeDbId);
private List<com.inteligr8.alfresco.asie.model.ShardInstance> findPossibleShardInstances(long nodeDbId) {
if (this.shardRegistry == null)
throw new UnsupportedOperationException("ACS instances without a sharding configuration are not yet implemented");
SearchParameters searchParams = new SearchParameters();
searchParams.setLanguage(SearchService.LANGUAGE_FTS_ALFRESCO);
searchParams.setQuery("@" + this.formatForFts(ContentModel.PROP_NODE_DBID) + ":" + nodeDbId);
List<com.inteligr8.alfresco.asie.model.ShardInstance> instances = new LinkedList<>();
List<ShardInstance> slicedInstances = this.shardRegistry.getIndexSlice(searchParams);
if (slicedInstances != null) {
this.logger.trace("Due to a sharding method, considering only applicable shards and their ASIE nodes: {}: {}", nodeDbId, slicedInstances);
for (ShardInstance instance : slicedInstances)
instances.add(this.toModel(instance));
} else {
for (Entry<Floc, Map<Shard, Set<ShardState>>> floc : this.shardRegistry.getFlocs().entrySet()) {
if (!floc.getKey().getStoreRefs().contains(StoreRef.STORE_REF_WORKSPACE_SPACESSTORE))
continue;
for (Entry<Shard, Set<ShardState>> shard : floc.getValue().entrySet()) {
for (ShardState shardState : shard.getValue())
instances.add(this.toModel(shardState.getShardInstance(), shardState));
}
}
this.logger.trace("Despite sharding, considering all shards and nodes: {}: {}", nodeDbId, instances);
}
return instances;
}
private com.inteligr8.alfresco.asie.model.ShardInstance toModel(ShardInstance instance) {
// get any random shardState
Floc floc = instance.getShard().getFloc();
Map<Shard, Set<ShardState>> shardsStates = this.shardRegistry.getFlocs().get(floc);
if (shardsStates == null)
throw new IllegalStateException();
Set<ShardState> shardStates = shardsStates.get(instance.getShard());
if (shardStates == null || shardStates.isEmpty())
throw new IllegalStateException();
ShardState anyShardState = shardStates.iterator().next();
return this.toModel(instance, anyShardState);
}
private com.inteligr8.alfresco.asie.model.ShardInstance toModel(ShardInstance instance, ShardState anyShardState) {
Floc floc = instance.getShard().getFloc();
ShardSet shardSet = ShardSet.from(floc, anyShardState);
SolrHost host = SolrHost.from(instance);
com.inteligr8.alfresco.asie.model.Shard shard = com.inteligr8.alfresco.asie.model.Shard.from(shardSet, instance.getShard().getInstance());
return com.inteligr8.alfresco.asie.model.ShardInstance.from(shard, host);
}
private String formatForFts(QName qname) {
return qname.toPrefixString(this.namespaceService).replace("-", "\\-").replace(":", "\\:");
}
}

View File

@@ -0,0 +1,309 @@
package com.inteligr8.alfresco.asie.service;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.alfresco.model.ContentModel;
import org.alfresco.model.RenditionModel;
import org.alfresco.service.cmr.repository.NodeRef;
import org.alfresco.service.cmr.repository.NodeRef.Status;
import org.alfresco.service.cmr.repository.NodeService;
import org.alfresco.service.cmr.repository.StoreRef;
import org.alfresco.service.cmr.search.QueryConsistency;
import org.alfresco.service.cmr.search.ResultSet;
import org.alfresco.service.cmr.search.SearchParameters;
import org.alfresco.service.cmr.search.SearchService;
import org.alfresco.service.namespace.NamespaceService;
import org.alfresco.service.namespace.QName;
import org.apache.commons.collections4.SetUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import com.inteligr8.alfresco.asie.model.ShardInstance;
import com.inteligr8.alfresco.asie.spi.ReconcileCallback;
import com.inteligr8.alfresco.asie.spi.ReindexCallback;
import com.inteligr8.alfresco.asie.util.CompositeFuture;
import com.inteligr8.alfresco.asie.util.ThrottledThreadPoolExecutor;
@Component
public class AcsReconcileService implements InitializingBean, DisposableBean {
private final Logger logger = LoggerFactory.getLogger(this.getClass());
private final Logger reconcileLogger = LoggerFactory.getLogger("inteligr8.asie.reconcile");
private final Set<QName> ignoreNodesWithAspects = SetUtils.unmodifiableSet(
RenditionModel.ASPECT_RENDITION,
RenditionModel.ASPECT_RENDITION2);
@Autowired
private NodeService nodeService;
@Autowired
private NamespaceService namespaceService;
@Autowired
private SearchService searchService;
@Autowired
private ReindexService reindexService;
@Value("${inteligr8.asie.reconciliation.nodesChunkSize:250}")
private int nodesChunkSize;
@Value("${inteligr8.asie.reconciliation.nodeTimeoutSeconds:10}")
private int nodeTimeoutSeconds;
@Value("${inteligr8.asie.reconciliation.concurrentQueueSize:64}")
private int concurrentQueueSize;
@Value("${inteligr8.asie.reconciliation.concurrency:2}")
private int concurrency;
private ThrottledThreadPoolExecutor executor;
@Override
public void afterPropertiesSet() {
this.executor = new ThrottledThreadPoolExecutor(this.concurrency, this.concurrency, this.concurrentQueueSize, 1L, TimeUnit.MINUTES, "solr-reconcile");
this.executor.prestartAllCoreThreads();
}
@Override
public void destroy() {
this.executor.shutdown();
}
/**
* This method reconciles the specified node range between ACS and Solr.
* The node range is specified using the ACS unique database identifiers.
* There is no other reasonably efficient attack vector. The callback
* handles all the return values. This is the synchronous alternative to
* the other `reconcile` method.
*
* There are two sets of parameters regarding timeouts. The queue timeouts
* are for how long the requesting thread should wait for a full queue to
* open up space for new re-index executions. The execution timeouts are
* for how long the execution should be allowed to take once dequeued.
* There is no timeout for how long the execution is queued.
*
* @param fromDbId A node database ID, inclusive.
* @param toDbId A node database ID, exclusive.
* @param reindexUnreconciled For nodes not found in Solr, attempt to re-index against all applicable Solr instances.
* @param callback A callback to process multiple returned values from the re-index.
* @param queueTimeout A timeout for how long the calling thread should wait for space on the queue.
* @param queueUnit The time units for the `queueTimeout`.
* @param execTimeout A timeout for the elapsed time the reindex execution should take when dequeued.
* @param execUnit The time units for the `execTimeout`.
* @throws TimeoutException Either the queue or execution timeout lapsed.
* @throws InterruptedException The re-index was interrupted (server shutdown).
*/
public void reconcile(
long fromDbId, long toDbId, Integer nodesChunkSize,
boolean reindexUnreconciled,
ReconcileCallback callback,
long queueTimeout, TimeUnit queueUnit,
long execTimeout, TimeUnit execUnit) throws InterruptedException, TimeoutException {
if (nodesChunkSize == null)
nodesChunkSize = this.nodesChunkSize;
if (this.logger.isTraceEnabled())
this.logger.trace("reconcile({}, {}, {}, {}, {}, {})", fromDbId, toDbId, nodesChunkSize, reindexUnreconciled, queueUnit.toMillis(queueTimeout), execUnit.toMillis(execTimeout));
CompositeFuture<Void> future = new CompositeFuture<>();
for (long startDbId = fromDbId; startDbId < toDbId; startDbId += nodesChunkSize) {
long endDbId = Math.min(toDbId, startDbId + nodesChunkSize);
future.combine(this.reconcileChunk(startDbId, endDbId, reindexUnreconciled, callback, queueTimeout, queueUnit, execTimeout, execUnit));
future.purge(true);
}
try {
future.get(execTimeout, execUnit);
} catch (ExecutionException ee) {
this.logger.error("Reconciliation thread failed: " + ee.getMessage(), ee);
}
}
public Future<Void> reconcile(
long fromDbId, long toDbId, Integer nodesChunkSize,
boolean reindexUnreconciled,
ReconcileCallback callback) throws InterruptedException {
if (nodesChunkSize == null)
nodesChunkSize = this.nodesChunkSize;
this.logger.trace("reconcile({}, {}, {}, {})", fromDbId, toDbId, nodesChunkSize, reindexUnreconciled);
CompositeFuture<Void> future = new CompositeFuture<>();
try {
for (long startDbId = fromDbId; startDbId < toDbId; startDbId += nodesChunkSize) {
long endDbId = Math.min(toDbId, startDbId + nodesChunkSize);
future.combine(this.reconcileChunk(startDbId, endDbId, reindexUnreconciled, callback, -1L, null, -1L, null));
future.purge(true);
}
} catch (TimeoutException te) {
throw new RuntimeException("This should never happen: " + te.getMessage(), te);
}
return future;
}
protected Future<Void> reconcileChunk(
long fromDbId, long toDbId,
boolean reindexUnreconciled,
ReconcileCallback callback,
long queueTimeout, TimeUnit queueUnit,
long execTimeout, TimeUnit execUnit) throws InterruptedException, TimeoutException {
if (this.logger.isTraceEnabled())
this.logger.trace("reconcileChunk({}, {}, {}, {}, {})", fromDbId, toDbId, reindexUnreconciled, queueUnit.toMillis(queueTimeout), execUnit.toMillis(execTimeout));
int dbIdCount = (int) (toDbId - fromDbId);
SearchParameters searchParams = new SearchParameters();
searchParams.addStore(StoreRef.STORE_REF_WORKSPACE_SPACESSTORE);
searchParams.setQueryConsistency(QueryConsistency.EVENTUAL); // force Solr
searchParams.setLanguage(SearchService.LANGUAGE_FTS_ALFRESCO);
searchParams.setQuery("@" + this.formatForFts(ContentModel.PROP_NODE_DBID) + ":[" + fromDbId + " TO " + toDbId + ">");
searchParams.setMaxItems(dbIdCount);
searchParams.setBulkFetchEnabled(false);
searchParams.setIncludeMetadata(false);
// `null`: unknown; or does not exist in DB
// `true`: exists in DB and Solr
// `false`: exists in DB, but not Solr
NodeRef[] nodeRefs = new NodeRef[dbIdCount];
this.logger.trace("Querying for nodes in chunk: {}", searchParams.getQuery());
ResultSet nodes = this.searchService.query(searchParams);
this.logger.debug("Found {} of {} possible nodes in chunk: {}-{}", nodes.getNumberFound(), dbIdCount, fromDbId, toDbId);
for (NodeRef nodeRef : nodes.getNodeRefs()) {
Status nodeStatus = this.nodeService.getNodeStatus(nodeRef);
long nodeDbId = nodeStatus.getDbId();
if (nodeDbId < fromDbId || nodeDbId >= toDbId) {
this.logger.warn("An unexpected DB ID was included in the result set; ignoring: {} != [{}, {})", nodeDbId, fromDbId, toDbId);
continue;
}
int dbIdIndex = (int) (nodeDbId - fromDbId);
nodeRefs[dbIdIndex] = nodeRef;
}
CompositeFuture<Void> future = new CompositeFuture<>();
for (long _nodeDbId = fromDbId; _nodeDbId < toDbId; _nodeDbId++) {
final long nodeDbId = _nodeDbId;
this.logger.trace("Attempting to reconcile ACS node: {}", nodeDbId);
final int dbIdIndex = (int) (nodeDbId - fromDbId);
if (nodeRefs[dbIdIndex] != null) {
this.logger.trace("A node in the DB is already indexed in Solr: {}: {}", nodeDbId, nodeRefs[dbIdIndex]);
this.reconcileLogger.info("RECONCILED: {} <=> {}", nodeDbId, nodeRefs[dbIdIndex]);
callback.reconciled(nodeDbId);
continue;
}
Callable<Void> callable = new Callable<Void>() {
@Override
public Void call() throws InterruptedException, TimeoutException {
reconcile(nodeDbId, reindexUnreconciled, callback, execTimeout, execUnit);
return null;
}
};
if (queueTimeout < 0L) {
future.combine(this.executor.submit(callable, -1L, null));
} else {
future.combine(this.executor.submit(callable, queueTimeout, queueUnit));
}
}
return future;
}
public void reconcile(long nodeDbId,
boolean reindexUnreconciled,
ReconcileCallback callback,
long execTimeout, TimeUnit execUnit) throws InterruptedException, TimeoutException {
NodeRef nodeRef = this.nodeService.getNodeRef(nodeDbId);
if (nodeRef == null) {
this.logger.trace("No such ACS node: {}; skipping ...", nodeDbId);
return;
}
if (!StoreRef.STORE_REF_WORKSPACE_SPACESSTORE.equals(nodeRef.getStoreRef())) {
this.logger.trace("A deliberately ignored store in the DB is not indexed in Solr: {}: {}", nodeDbId, nodeRef);
return;
}
Set<QName> aspects = this.nodeService.getAspects(nodeRef);
aspects.retainAll(this.ignoreNodesWithAspects);
if (!aspects.isEmpty()) {
this.logger.trace("A deliberately ignored node in the DB is not indexed in Solr: {}: {}: {}", nodeDbId, nodeRef, aspects);
return;
}
if (!reindexUnreconciled) {
this.logger.debug("A node in the DB is not indexed in Solr: {}: {}", nodeDbId, nodeRef);
this.reconcileLogger.info("UNRECONCILED: {} <=> {}", nodeDbId, nodeRef);
callback.unreconciled(nodeDbId);
} else {
logger.debug("A node in the DB is not indexed in Solr; attempt to reindex: {}: {}", nodeDbId, nodeRef);
this.reindex(nodeDbId, nodeRef, callback, execTimeout, execUnit);
}
}
public void reindex(long nodeDbId, NodeRef nodeRef,
ReconcileCallback callback,
long execTimeout, TimeUnit execUnit) throws InterruptedException, TimeoutException {
Set<ShardInstance> syncHosts = new HashSet<>();
Set<ShardInstance> asyncHosts = new HashSet<>();
Map<ShardInstance, String> errorHosts = new HashMap<>();
ReindexCallback reindexCallback = new ReindexCallback() {
@Override
public void success(ShardInstance instance) {
reconcileLogger.info("REINDEXED: {} <=> {}", nodeDbId, nodeRef);
syncHosts.add(instance);
}
@Override
public void scheduled(ShardInstance instance) {
reconcileLogger.info("REINDEXING: {} <=> {}", nodeDbId, nodeRef);
asyncHosts.add(instance);
}
@Override
public void error(ShardInstance instance, String message) {
reconcileLogger.info("UNINDEXED: {} <=> {}", nodeDbId, nodeRef);
errorHosts.put(instance, message);
}
};
try {
if (execTimeout < 0L) {
this.reindexService.reindex(nodeDbId, reindexCallback).get();
} else {
this.reindexService.reindex(nodeDbId, reindexCallback).get(execTimeout, execUnit);
}
} catch (ExecutionException ee) {
throw new RuntimeException("An unexpected exception occurred: " + ee.getMessage(), ee);
}
if (callback != null)
callback.processed(nodeDbId, syncHosts, asyncHosts, errorHosts);
}
private String formatForFts(QName qname) {
return qname.toPrefixString(this.namespaceService).replace("-", "\\-").replace(":", "\\:");
}
}

View File

@@ -0,0 +1,118 @@
package com.inteligr8.alfresco.asie.service;
import java.io.IOException;
import java.net.URL;
import org.alfresco.repo.index.shard.ShardInstance;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import com.inteligr8.alfresco.asie.model.SolrHost;
import com.inteligr8.rs.AuthorizationFilter;
import com.inteligr8.rs.Client;
import com.inteligr8.rs.ClientCxfConfiguration;
import com.inteligr8.rs.ClientCxfImpl;
import jakarta.ws.rs.client.ClientRequestContext;
@Component
public class ApiService implements InitializingBean {
public static final String SOLR_CORE = "alfresco";
private final Logger logger = LoggerFactory.getLogger(this.getClass());
@Value("${solr.secureComms}")
private String solrSecureComms;
@Value("${solr.port}")
private int solrPort;
@Value("${solr.port.ssl}")
private int solrSslPort;
@Value("${solr.sharedSecret.header}")
private String solrSharedSecretHeader;
@Value("${solr.sharedSecret}")
private String solrSharedSecret;
@Value("${inteligr8.asie.basePath}")
private String solrBaseUrl;
@Value("${inteligr8.asie.reconciliation.nodesChunkSize:250}")
private int nodesChunkSize;
@Override
public void afterPropertiesSet() throws Exception {
this.solrSharedSecret = StringUtils.trimToNull(this.solrSharedSecret);
}
public <T> T createApi(String hostname, int port, Class<T> apiClass) {
String solrBaseUrl = this.formulateSolrBaseUrl(hostname, port);
this.logger.trace("Using Solr base URL: {}", solrBaseUrl);
return this.createApi(solrBaseUrl, apiClass);
}
public <T> T createApi(ShardInstance instance, Class<T> apiClass) {
URL url = SolrHost.from(instance).toUrl("http");
return this.createApi(url.toString(), apiClass);
}
public <T> T createApi(String solrBaseUrl, Class<T> apiClass) {
Client solrClient = this.createClient(solrBaseUrl);
return this.getApi(solrClient, apiClass);
}
public <T> T getApi(Client solrClient, Class<T> apiClass) {
return solrClient.getApi(apiClass);
}
public boolean isSecure() {
return "https".equals(this.solrSecureComms);
}
public int getDefaultSolrPort() {
return this.isSecure() ? this.solrSslPort : this.solrPort;
}
public String formulateSolrBaseUrl(String hostname, Integer port) {
boolean isSsl = "https".equals(this.solrSecureComms);
StringBuilder baseUrl = new StringBuilder(isSsl ? "https" : "http").append("://").append(hostname);
baseUrl.append(':').append(port == null ? (isSsl ? this.solrSslPort : this.solrPort) : port);
baseUrl.append(this.solrBaseUrl);
return baseUrl.toString();
}
public Client createClient(final String baseUrl) {
ClientCxfImpl client = new ClientCxfImpl(new ClientCxfConfiguration() {
@Override
public String getBaseUrl() {
return baseUrl.toString();
}
@Override
public AuthorizationFilter createAuthorizationFilter() {
return solrSharedSecret == null ? null : new AuthorizationFilter() {
@Override
public void filter(ClientRequestContext requestContext) throws IOException {
logger.trace("Adding authorization headers for ASIE shared auth: {}", solrSharedSecretHeader);
requestContext.getHeaders().putSingle(solrSharedSecretHeader, solrSharedSecret);
}
};
}
@Override
public boolean isDefaultBusEnabled() {
return false;
}
});
client.register();
return client;
}
}

View File

@@ -0,0 +1,138 @@
package com.inteligr8.alfresco.asie.service;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.TimeUnit;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import com.google.common.cache.RemovalListener;
import com.google.common.cache.RemovalNotification;
import com.inteligr8.alfresco.asie.util.ThrottledThreadPoolExecutor;
/**
* This class manages the instantiation and shutdown of `ExecutorService`
* instances that are "rarely" used.
*
* The default implementation of each `ExecutorService` do not support 0 core
* threads unless you want undesirable effects. So they hold threads and could
* hold them for weeks without being used.
*
* We will shutdown and dereference the whole `ExecutorService` when there are
* no other hard references and after the `keepAliveTime` expires.
*/
@Component
public class ExecutorManager implements InitializingBean, DisposableBean, RemovalListener<String, ExecutorService> {
@Value("${inteligr8.asie.executors.expireTimeInMinutes:30}")
private int expireTimeInMinutes;
private Cache<String, ExecutorService> refCache;
private Cache<String, ExecutorService> expiringCache;
@Override
public void afterPropertiesSet() throws Exception {
// a weak value happens when the executor is no longer referenced
// the possible references are by the caller temporarily using and the `expiringCache` (below; so it expired)
// this keeps the pool from being shutdown after it expires if the caller is still referencing it
// ultimately, if it is cached, it will be in this cache and MAY be in the `expiringCache`.
this.refCache = CacheBuilder.newBuilder()
.initialCapacity(8)
.weakValues()
.removalListener(this)
.build();
this.expiringCache = CacheBuilder.newBuilder()
.initialCapacity(8)
.expireAfterAccess(this.expireTimeInMinutes, TimeUnit.MINUTES)
.build();
}
@Override
public void destroy() throws Exception {
this.refCache.invalidateAll();
this.refCache.cleanUp();
this.expiringCache.invalidateAll();
this.expiringCache.cleanUp();
}
@Override
public void onRemoval(RemovalNotification<String, ExecutorService> notification) {
notification.getValue().shutdown();
}
public ThrottledThreadPoolExecutor createThrottled(
String name,
int coreThreadPoolSize,
int maximumThreadPoolSize,
int maximumQueueSize,
long keepAliveTime,
TimeUnit unit) {
return this.createThrottled(name, coreThreadPoolSize, maximumThreadPoolSize, maximumQueueSize, keepAliveTime, unit, null);
}
public ThrottledThreadPoolExecutor createThrottled(
final String name,
final int coreThreadPoolSize,
final int maximumThreadPoolSize,
final int maximumQueueSize,
final long keepAliveTime,
final TimeUnit unit,
final RejectedExecutionHandler rejectedExecutionHandler) {
try {
// if it is already cached, reuse the cache; otherwise create one
final ExecutorService executor = this.refCache.get(name, new Callable<ThrottledThreadPoolExecutor>() {
@Override
public ThrottledThreadPoolExecutor call() {
ThrottledThreadPoolExecutor executor = null;
if (rejectedExecutionHandler == null) {
executor = new ThrottledThreadPoolExecutor(coreThreadPoolSize, maximumThreadPoolSize, maximumQueueSize,
keepAliveTime, unit,
name);
} else {
executor = new ThrottledThreadPoolExecutor(coreThreadPoolSize, maximumThreadPoolSize, maximumQueueSize,
keepAliveTime, unit,
name,
rejectedExecutionHandler);
}
executor.prestartAllCoreThreads();
return executor;
}
});
return (ThrottledThreadPoolExecutor) this.expiringCache.get(name, new Callable<ExecutorService>() {
@Override
public ExecutorService call() throws Exception {
return executor;
}
});
} catch (ExecutionException ee) {
throw new RuntimeException("This should never happen", ee);
}
}
public ExecutorService get(String name) {
// grab from the expiring cache first, so we can
ExecutorService executor = this.expiringCache.getIfPresent(name);
if (executor != null)
return executor;
executor = this.refCache.getIfPresent(name);
if (executor == null)
return null;
// the executor expired, but it was still referenced by the caller
// re-cache it
this.expiringCache.put(name, executor);
return executor;
}
}

View File

@@ -0,0 +1,43 @@
package com.inteligr8.alfresco.asie.service;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.springframework.stereotype.Component;
import com.inteligr8.alfresco.asie.api.CoreAdminApi;
import com.inteligr8.alfresco.asie.model.ActionCoreResponse;
import com.inteligr8.alfresco.asie.model.core.FixAction;
import com.inteligr8.alfresco.asie.model.core.FixRequest;
import com.inteligr8.alfresco.asie.spi.ActionCallback;
@Component
public class FixService extends AbstractActionService {
@Override
protected String getActionName() {
return "fix";
}
@Override
protected String getThreadNamePrefix() {
return "solr-fix";
}
@Override
protected ActionCoreResponse<FixAction> execute(CoreAdminApi api, String core) {
FixRequest apiRequest = new FixRequest().withCore(core);
return api.fix(apiRequest);
}
public Future<Void> fix(ActionCallback callback) throws InterruptedException {
return super.action(callback);
}
public void fix(ActionCallback callback, long fullQueueTimeout, TimeUnit fullQueueUnit,
long execTimeout, TimeUnit execUnit) throws TimeoutException, InterruptedException {
super.action(callback, fullQueueTimeout, fullQueueUnit, execTimeout, execUnit);
}
}

View File

@@ -0,0 +1,43 @@
package com.inteligr8.alfresco.asie.service;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.springframework.stereotype.Component;
import com.inteligr8.alfresco.asie.api.CoreAdminApi;
import com.inteligr8.alfresco.asie.model.ActionCoreResponse;
import com.inteligr8.alfresco.asie.model.core.PurgeRequest;
import com.inteligr8.alfresco.asie.spi.ActionCallback;
import com.inteligr8.solr.model.Action;
@Component
public class PurgeService extends AbstractNodeActionService {
@Override
protected String getActionName() {
return "purge";
}
@Override
protected String getThreadNamePrefix() {
return "solr-purge";
}
@Override
protected ActionCoreResponse<Action> execute(CoreAdminApi api, String core, long nodeDbId) {
PurgeRequest apiRequest = new PurgeRequest().withCore(core).withNodeId(nodeDbId);
return api.purge(apiRequest);
}
public Future<Void> purge(long nodeDbId, ActionCallback callback) throws InterruptedException {
return super.action(nodeDbId, callback);
}
public void purge(long nodeDbId, ActionCallback callback, long fullQueueTimeout, TimeUnit fullQueueUnit,
long execTimeout, TimeUnit execUnit) throws TimeoutException, InterruptedException {
super.action(nodeDbId, callback, fullQueueTimeout, fullQueueUnit, execTimeout, execUnit);
}
}

View File

@@ -0,0 +1,43 @@
package com.inteligr8.alfresco.asie.service;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.springframework.stereotype.Component;
import com.inteligr8.alfresco.asie.api.CoreAdminApi;
import com.inteligr8.alfresco.asie.model.ActionCoreResponse;
import com.inteligr8.alfresco.asie.model.core.ReindexRequest;
import com.inteligr8.alfresco.asie.spi.ActionCallback;
import com.inteligr8.solr.model.Action;
@Component
public class ReindexService extends AbstractNodeActionService {
@Override
protected String getActionName() {
return "re-index";
}
@Override
protected String getThreadNamePrefix() {
return "solr-reindex";
}
@Override
protected ActionCoreResponse<Action> execute(CoreAdminApi api, String core, long nodeDbId) {
ReindexRequest apiRequest = new ReindexRequest().withCore(core).withNodeId(nodeDbId);
return api.reindex(apiRequest);
}
public Future<Void> reindex(long nodeDbId, ActionCallback callback) throws InterruptedException {
return super.action(nodeDbId, callback);
}
public void reindex(long nodeDbId, ActionCallback callback, long fullQueueTimeout, TimeUnit fullQueueUnit,
long execTimeout, TimeUnit execUnit) throws TimeoutException, InterruptedException {
super.action(nodeDbId, callback, fullQueueTimeout, fullQueueUnit, execTimeout, execUnit);
}
}

View File

@@ -0,0 +1,43 @@
package com.inteligr8.alfresco.asie.service;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.springframework.stereotype.Component;
import com.inteligr8.alfresco.asie.api.CoreAdminApi;
import com.inteligr8.alfresco.asie.model.ActionCoreResponse;
import com.inteligr8.alfresco.asie.model.core.RetryAction;
import com.inteligr8.alfresco.asie.model.core.RetryRequest;
import com.inteligr8.alfresco.asie.spi.ActionCallback;
@Component
public class RetryService extends AbstractActionService {
@Override
protected String getActionName() {
return "retry";
}
@Override
protected String getThreadNamePrefix() {
return "solr-retry";
}
@Override
protected ActionCoreResponse<RetryAction> execute(CoreAdminApi api, String core) {
RetryRequest apiRequest = new RetryRequest().withCore(core);
return api.retry(apiRequest);
}
public Future<Void> retry(ActionCallback callback) throws InterruptedException {
return super.action(callback);
}
public void retry(ActionCallback callback, long fullQueueTimeout, TimeUnit fullQueueUnit,
long execTimeout, TimeUnit execUnit) throws TimeoutException, InterruptedException {
super.action(callback, fullQueueTimeout, fullQueueUnit, execTimeout, execUnit);
}
}

View File

@@ -1,7 +1,5 @@
package com.inteligr8.alfresco.asie.service;
import java.io.Serializable;
import org.alfresco.service.cmr.attributes.AttributeService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -11,6 +9,7 @@ import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import com.inteligr8.alfresco.asie.Constants;
import com.inteligr8.alfresco.asie.model.PersistedNode;
import com.inteligr8.alfresco.asie.model.ShardSet;
import com.inteligr8.alfresco.asie.model.SolrHost;
@@ -31,13 +30,13 @@ public class ShardBackupService implements com.inteligr8.alfresco.asie.spi.Shard
String shardKey = shardSet.getCore() + "-" + shardId;
PersistedNode backupNode = (PersistedNode) this.attributeService.getAttribute(Constants.ATTR_ASIE, ATTR_BACKUP_NODE, shardKey);
this.logger.debug("Found backup node: {}", backupNode);
logger.debug("Found backup node: {}", backupNode);
if (backupNode == null || backupNode.isExpired()) {
backupNode = new PersistedNode(node);
backupNode = new PersistedNode(node, this.persistTimeMinutes);
this.attributeService.setAttribute(backupNode, Constants.ATTR_ASIE, ATTR_BACKUP_NODE, shardKey);
}
return backupNode.getNode();
}
@@ -49,38 +48,5 @@ public class ShardBackupService implements com.inteligr8.alfresco.asie.spi.Shard
String shardKey = shardSet.getCore() + "-" + shardId;
this.attributeService.removeAttribute(Constants.ATTR_ASIE, ATTR_BACKUP_NODE, shardKey);
}
private class PersistedNode implements Serializable {
private static final long serialVersionUID = 4105196543023419818L;
private final SolrHost node;
private long expireTimeMillis;
PersistedNode(SolrHost node) {
this.node = node;
this.reset();
}
void reset() {
this.expireTimeMillis = System.currentTimeMillis() + persistTimeMinutes * 60L * 1000L;
}
boolean isExpired() {
return this.expireTimeMillis < System.currentTimeMillis();
}
SolrHost getNode() {
return this.node;
}
@Override
public String toString() {
return "node: " + this.node + "; expires in: " + (System.currentTimeMillis() - this.expireTimeMillis) + " ms";
}
}
}

View File

@@ -0,0 +1,15 @@
package com.inteligr8.alfresco.asie.spi;
import com.inteligr8.alfresco.asie.model.ShardInstance;
public interface ActionCallback {
void success(ShardInstance instance);
void scheduled(ShardInstance instance);
void error(ShardInstance instance, String message);
void unknownResult(ShardInstance instance);
}

View File

@@ -0,0 +1,19 @@
package com.inteligr8.alfresco.asie.spi;
import java.util.Map;
import java.util.Set;
import com.inteligr8.alfresco.asie.model.ShardInstance;
public interface ReconcileCallback {
void reconciled(long nodeDbId);
void unreconciled(long nodeDbId);
void processed(long nodeDbId,
Set<ShardInstance> instsReconciled,
Set<ShardInstance> instsReconciling,
Map<ShardInstance, String> instsErrorMessages);
}

View File

@@ -0,0 +1,12 @@
package com.inteligr8.alfresco.asie.spi;
import com.inteligr8.alfresco.asie.model.ShardInstance;
public interface ReindexCallback extends ActionCallback {
@Override
default void unknownResult(ShardInstance instance) {
throw new IllegalStateException();
}
}

View File

@@ -0,0 +1,150 @@
package com.inteligr8.alfresco.asie.util;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.apache.commons.lang3.tuple.MutablePair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class CompositeFuture<T> implements Future<T> {
private final Logger logger = LoggerFactory.getLogger(this.getClass());
private final Collection<Future<T>> futures = new ConcurrentLinkedQueue<>();
public MutablePair<Integer, Integer> counts() {
MutablePair<Integer, Integer> counts = MutablePair.of(0, 0);
for (Future<T> future : this.futures) {
if (future.isDone()) {
counts.setRight(counts.getRight().intValue() + 1);
} else {
counts.setLeft(counts.getLeft().intValue() + 1);
}
}
return counts;
}
public int countIncomplete() {
return this.counts().getLeft().intValue();
}
public int countCompleted() {
return this.counts().getRight().intValue();
}
public void combine(Future<T> future) {
this.futures.add(future);
}
public void combine(Collection<Future<T>> futures) {
this.futures.addAll(futures);
}
@Override
public boolean cancel(boolean mayInterruptIfRunning) {
boolean cancelled = true;
for (Future<T> future : this.futures)
if (!future.cancel(mayInterruptIfRunning))
cancelled = false;
return cancelled;
}
public List<T> getList() throws InterruptedException, ExecutionException {
List<T> results = new ArrayList<>(this.futures.size());
for (Future<T> future : this.futures)
results.add(future.get());
return results;
}
@Override
public T get() throws InterruptedException, ExecutionException {
this.getList();
return null;
}
public List<T> getList(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
long expireTimeMillis = System.currentTimeMillis() + unit.toMillis(timeout);
List<T> results = new ArrayList<>(this.futures.size());
for (Future<T> future : this.futures) {
if (future instanceof RunnableFuture<?>) {
this.logger.debug("Waiting {} ms since the start of the exectuion of the future to complete", unit.toMillis(timeout));
results.add(((RunnableFuture<T>) future).get(timeout, unit));
} else {
long remainingTimeMillis = expireTimeMillis - System.currentTimeMillis();
this.logger.debug("Waiting {} ms for the future to complete", remainingTimeMillis);
results.add(future.get(remainingTimeMillis, TimeUnit.MILLISECONDS));
}
}
return results;
}
@Override
public T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
this.getList(timeout, unit);
return null;
}
@Override
public boolean isCancelled() {
for (Future<T> future : this.futures)
if (!future.isCancelled())
return false;
return true;
}
@Override
public boolean isDone() {
for (Future<T> future : this.futures)
if (!future.isDone())
return false;
return true;
}
/**
* Remove any futures that are done (or cancelled).
*
* @param includeCancelled `true` to purge cancelled futures; `false` to purge only completed futures
*/
public void purge(boolean includeCancelled) {
List<CompositeFuture<?>> cfutures = new LinkedList<>();
int removedCancelled = 0;
int removedDone = 0;
Iterator<Future<T>> i = this.futures.iterator();
while (i.hasNext()) {
Future<T> future = i.next();
if (future.isCancelled()) {
if (includeCancelled) {
removedCancelled++;
i.remove();
}
} else if (future.isDone()) {
removedDone++;
i.remove();
} else if (future instanceof CompositeFuture<?>) {
cfutures.add((CompositeFuture<?>) future);
}
}
this.logger.debug("Purged {} cancelled and {} completed futures", removedCancelled, removedDone);
for (CompositeFuture<?> cfuture : cfutures)
cfuture.purge(includeCancelled);
}
}

View File

@@ -0,0 +1,43 @@
package com.inteligr8.alfresco.asie.util;
public abstract class ElapsedInterruptableRunnable extends InterruptableRunnable {
private final Object runLock = new Object();
private boolean runOnce = true;
private boolean run = false;
private Long executionStartTimeMillis;
private Long executionEndTimeMillis;
public void setRunOnce(boolean runOnce) {
this.runOnce = runOnce;
}
@Override
public final void runInterruptable() throws InterruptedException {
synchronized (this.runLock) {
if (this.runOnce && this.run)
return;
this.run = true;
}
this.executionStartTimeMillis = System.currentTimeMillis();
try {
this.runElapsed();
} finally {
this.executionEndTimeMillis = System.currentTimeMillis();
}
}
protected abstract void runElapsed() throws InterruptedException;
public Long computeElapsedExecutionMillis() {
if (this.executionEndTimeMillis != null) {
return this.executionEndTimeMillis - this.executionStartTimeMillis;
} else if (this.executionStartTimeMillis != null) {
return System.currentTimeMillis() - this.executionStartTimeMillis;
} else {
return null;
}
}
}

View File

@@ -0,0 +1,75 @@
package com.inteligr8.alfresco.asie.util;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public abstract class InterruptableRunnable implements Runnable {
private final Logger logger = LoggerFactory.getLogger(this.getClass());
private final Object threadLock = new Object();
private boolean run = false;
private Thread runThread = null;
private boolean completed = false;
private boolean interrupted = false;
@Override
public final void run() {
synchronized (this.threadLock) {
this.run = true;
this.runThread = Thread.currentThread();
}
try {
this.runInterruptable();
this.completed = true;
} catch (InterruptedException ie) {
this.logger.debug("Runnable interrupted");
this.interrupted = true;
this.interrupted(ie);
} finally {
this.runThread = null;
}
}
protected abstract void runInterruptable() throws InterruptedException;
public boolean interrupt() {
synchronized (this.threadLock) {
if (this.runThread == null)
return false;
this.logger.trace("Runnable interrupting ...");
this.runThread.interrupt();
}
return true;
}
protected void interrupted(InterruptedException ie) {
}
public boolean isQueued() {
return !this.run;
}
public boolean isInterrupted() {
return this.interrupted;
}
public boolean isCompleted() {
return this.completed;
}
public boolean isFailed() {
synchronized (this.threadLock) {
return this.run && !this.completed && !this.interrupted && this.runThread == null;
}
}
public boolean isDone() {
synchronized (this.threadLock) {
return this.run && this.runThread == null;
}
}
}

View File

@@ -0,0 +1,72 @@
package com.inteligr8.alfresco.asie.util;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class RunnableFuture<T> implements Future<T> {
private final Logger logger = LoggerFactory.getLogger(this.getClass());
private final ThreadPoolExecutor executor;
private final WaitableRunnable runnable;
public RunnableFuture(ThreadPoolExecutor executor, WaitableRunnable runnable) {
this.executor = executor;
this.runnable = runnable;
}
@Override
public boolean cancel(boolean mayInterruptIfRunning) {
if (this.executor.remove(this.runnable)) {
this.logger.debug("Cancelled runnable by removing from queue");
return true;
} else if (mayInterruptIfRunning && this.runnable.interrupt()) {
this.logger.debug("Cancelled runnable by interrupting it");
return true;
}
return false;
}
/**
* This method will not never timeout and will only return or throw when
* the runnable is complete or interrupted.
*
* @return Always `null`.
*/
@Override
public T get() throws InterruptedException {
this.runnable.waitUntilDone();
return null;
}
/**
* This method will timeout in the specified amount of time after the
* execution commences. This will wait indefinitely as the execution is
* queued. Once it is in the executing state, the clock starts. This
* alters the default behavior of the `Future` interface.
*
* @param timeout A positive integer representing a period of time.
* @param unit The time unit of the `timeout` parameter.
* @return Always `null`.
*/
@Override
public T get(long timeout, TimeUnit unit) throws InterruptedException, TimeoutException {
this.runnable.waitUntilExecutionElapsed(timeout, unit);
return null;
}
@Override
public boolean isCancelled() {
return this.runnable.isInterrupted();
}
@Override
public boolean isDone() {
return this.runnable.isDone();
}
}

View File

@@ -0,0 +1,96 @@
package com.inteligr8.alfresco.asie.util;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
public class ThrottledThreadPoolExecutor extends ThreadPoolExecutor {
public ThrottledThreadPoolExecutor(
int coreThreadPoolSize,
int maximumThreadPoolSize,
int maximumQueueSize,
long keepAliveTime,
TimeUnit unit,
String threadNamePrefix) {
super(coreThreadPoolSize, maximumThreadPoolSize, keepAliveTime, unit,
new ArrayBlockingQueue<>(maximumQueueSize),
new ThreadFactoryBuilder()
.setNameFormat(threadNamePrefix + "-%d")
.build());
}
public ThrottledThreadPoolExecutor(
int coreThreadPoolSize,
int maximumThreadPoolSize,
int maximumQueueSize,
long keepAliveTime,
TimeUnit unit,
String threadNamePrefix,
RejectedExecutionHandler rejectedExecutionHandler) {
super(coreThreadPoolSize, maximumThreadPoolSize, keepAliveTime, unit,
new ArrayBlockingQueue<>(maximumQueueSize),
new ThreadFactoryBuilder()
.setNameFormat(threadNamePrefix + "-%d")
.build(),
rejectedExecutionHandler);
}
/**
* @param timeout Negative to not wait
*/
public <T> RunnableFuture<T> submit(Callable<T> task, long timeout, TimeUnit unit) throws InterruptedException, TimeoutException {
WaitableRunnable wrunnable = new WaitableRunnable() {
@Override
protected void runWaitable() throws InterruptedException {
try {
task.call();
} catch (InterruptedException ie) {
throw ie;
} catch (Exception e) {
throw new RuntimeException(e);
}
}
};
return new RunnableFuture<T>(this, this.submit(wrunnable, timeout, unit));
}
/**
* @param timeout Negative to not wait
*/
public RunnableFuture<?> submit(Runnable runnable, long timeout, TimeUnit unit) throws InterruptedException, TimeoutException {
WaitableRunnable wrunnable = new WaitableRunnable() {
@Override
protected void runWaitable() {
runnable.run();
}
};
return new RunnableFuture<Void>(this, this.submit(wrunnable, timeout, unit));
}
private WaitableRunnable submit(WaitableRunnable runnable, long throttlingBlockTimeout, TimeUnit throttlingBlockUnit) throws InterruptedException, TimeoutException {
// if no core threads are running, the queue won't be monitored for runnables
this.prestartAllCoreThreads();
if (throttlingBlockTimeout < 0L) {
this.getQueue().put(runnable);
} else {
if (!this.getQueue().offer(runnable, throttlingBlockTimeout, throttlingBlockUnit))
throw new TimeoutException();
}
return runnable;
}
}

View File

@@ -0,0 +1,60 @@
package com.inteligr8.alfresco.asie.util;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
public abstract class WaitableRunnable extends ElapsedInterruptableRunnable {
private final Object lock = new Object();
private boolean done = false;
@Override
public final void runElapsed() throws InterruptedException {
this.done = false;
try {
this.runWaitable();
} finally {
synchronized (this.lock) {
this.done = true;
this.lock.notifyAll();
}
}
}
protected abstract void runWaitable() throws InterruptedException;
public void waitUntilDone() throws InterruptedException {
synchronized (this.lock) {
if (!this.done && !this.isDone())
this.lock.wait();
}
}
public void waitUntilDone(long timeout, TimeUnit unit) throws InterruptedException, TimeoutException {
synchronized (this.lock) {
if (!this.done && !this.isDone()) {
long waitTime = unit.toMillis(timeout);
long startTime = System.currentTimeMillis();
this.lock.wait(waitTime);
if (System.currentTimeMillis() - startTime >= waitTime)
throw new TimeoutException();
}
}
}
public void waitUntilExecutionElapsed(long timeout, TimeUnit unit) throws InterruptedException, TimeoutException {
synchronized (this.lock) {
if (!this.done && !this.isDone()) {
while (this.isQueued())
Thread.sleep(50L);
Long elapsedExecutionMillis = this.computeElapsedExecutionMillis();
long waitTime = unit.toMillis(timeout) - elapsedExecutionMillis;
long startTime = System.currentTimeMillis();
this.lock.wait(waitTime);
if (System.currentTimeMillis() - startTime >= waitTime)
throw new TimeoutException();
}
}
}
}

View File

@@ -38,6 +38,9 @@
<!-- Security -->
<authentication>none</authentication>
<!-- Transaction -->
<transaction>required</transaction>
<!-- Functionality -->
<cache>
<never>false</never>

View File

@@ -0,0 +1,46 @@
<webscript xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:noNamespaceSchemaLocation="https://bitbucket.org/!api/2.0/snippets/inteligr8/AzMgbp/80fdd26a6b3769a63cdc6b54bf1f39e378545cf7/files/snippet.txt">
<!-- Naming & Organization -->
<shortname>Fix ASIE Indexes</shortname>
<family>Inteligr8 ASIE</family>
<description><![CDATA[
<p>Issue a 'fix' command to the ASIE indexes.
This call will attempt to fix all Solr nodes.
The fix operation is asynchronous and could fail on any Solr node without notification.</p>
<p>The following response body should be expected in most cases (202 and 500 status codes):</p>
<pre>
{
"scheduled": [
"solrHostAsync:8983/solr",
...
],
"error": [
"solrHostThatFailed:8983/solr": {
"message": "string"
},
...
]
}
</pre>
<p>The following status codes should be expected:</p>
<dl>
<dt>202</dt>
<dd>Accepted</dd>
</dl>
]]></description>
<!-- Endpoint Configuration -->
<url>/inteligr8/asie/acs/fix</url>
<format default="json">any</format>
<!-- Security -->
<authentication>user</authentication>
<!-- Functionality -->
<cache>
<never>false</never>
<public>false</public>
</cache>
</webscript>

View File

@@ -34,6 +34,9 @@
<!-- Security -->
<authentication>none</authentication>
<!-- Transaction -->
<transaction>required</transaction>
<!-- Functionality -->
<cache>
<never>false</never>

View File

@@ -29,7 +29,10 @@
<url>/inteligr8/asie/node/{nodeEndpoint}</url>
<!-- Security -->
<authentication>admin</authentication>
<authentication>user</authentication>
<!-- Transaction -->
<transaction>required</transaction>
<!-- Functionality -->
<cache>

View File

@@ -58,7 +58,7 @@
<format default="json">any</format>
<!-- Security -->
<authentication>admin</authentication>
<authentication>user</authentication>
<!-- Functionality -->
<cache>

View File

@@ -31,7 +31,7 @@
<url>/inteligr8/asie/node/{nodeEndpoint}?coreName={coreName?}&amp;shardRange={shardRange?}&amp;template={template?}&amp;shardCount={shardCount?}&amp;nodeId={nodeId?}&amp;nodeCount={nodeCount?}</url>
<!-- Security -->
<authentication>admin</authentication>
<authentication>user</authentication>
<!-- Functionality -->
<cache>

View File

@@ -32,7 +32,10 @@
<url>/inteligr8/asie/node/{nodeEndpoint}/shard/{shardCore}/{shardId}</url>
<!-- Security -->
<authentication>admin</authentication>
<authentication>user</authentication>
<!-- Transaction -->
<transaction>required</transaction>
<!-- Functionality -->
<cache>

View File

@@ -30,7 +30,7 @@
<url>/inteligr8/asie/node/{nodeEndpoint}/shard/{shardCore}/{shardId}</url>
<!-- Security -->
<authentication>admin</authentication>
<authentication>user</authentication>
<!-- Functionality -->
<cache>

View File

@@ -30,7 +30,10 @@
<url>/inteligr8/asie/node/{nodeEndpoint}/shard/{shardCore}/{shardId}</url>
<!-- Security -->
<authentication>admin</authentication>
<authentication>user</authentication>
<!-- Transaction -->
<transaction>required</transaction>
<!-- Functionality -->
<cache>

View File

@@ -54,7 +54,7 @@
<format default="json">any</format>
<!-- Security -->
<authentication>admin</authentication>
<authentication>user</authentication>
<!-- Functionality -->
<cache>

View File

@@ -61,7 +61,7 @@
<format default="json">any</format>
<!-- Security -->
<authentication>admin</authentication>
<authentication>user</authentication>
<!-- Functionality -->
<cache>

View File

@@ -0,0 +1,63 @@
<webscript xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:noNamespaceSchemaLocation="https://bitbucket.org/!api/2.0/snippets/inteligr8/AzMgbp/80fdd26a6b3769a63cdc6b54bf1f39e378545cf7/files/snippet.txt">
<!-- Naming & Organization -->
<shortname>Purge ACS Node in ASIE Indexes</shortname>
<family>Inteligr8 ASIE</family>
<description><![CDATA[
<p>Purge the specified ACS node in the ASIE indexes.
This call will attempt to purge the ACS node on all applicable Solr nodes.
The purge operation could be synchronous or asynchronous and could fail on any Solr node.
If any Solr node failed synchronously in the execution, then expect a status code of 500.
The response body will still be identical to the 200/202 status codes.
If any Solr node is executing the purge asynchronously and there are no synchronous failures, then expect a status code of 202.</p>
<p>The following path parameters are supported:</p>
<dl>
<dt>nodeId</dt>
<dd>An ACS node ID.</dd>
</dl>
<p>The following response body should be expected in most cases (200, 202, and 500 status codes):</p>
<pre>
{
"nodeDbId": number,
"success": [
"solrHostSync:8983/solr",
...
],
"scheduled": [
"solrHostAsync:8983/solr",
...
],
"error": [
"solrHostThatFailed:8983/solr": {
"message": "string"
},
...
]
}
</pre>
<p>The following status codes should be expected:</p>
<dl>
<dt>200</dt>
<dd>OK</dd>
<dt>202</dt>
<dd>Accepted</dd>
<dt>400</dt>
<dd>The path or query parameters are invalid</dd>
</dl>
]]></description>
<!-- Endpoint Configuration -->
<url>/inteligr8/asie/acs/node/{nodeId}/purge</url>
<format default="json">any</format>
<!-- Security -->
<authentication>user</authentication>
<!-- Functionality -->
<cache>
<never>false</never>
<public>false</public>
</cache>
</webscript>

View File

@@ -0,0 +1,73 @@
<webscript xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:noNamespaceSchemaLocation="https://bitbucket.org/!api/2.0/snippets/inteligr8/AzMgbp/80fdd26a6b3769a63cdc6b54bf1f39e378545cf7/files/snippet.txt">
<!-- Naming & Organization -->
<shortname>Reconcile ACS Nodes against ASIE Indexes</shortname>
<family>Inteligr8 ASIE</family>
<description><![CDATA[
<p>Reconcile the ACS node range against the ASIE indexes.
This call will loop through the specified ACS node range, reconciling each against ASIE indexes.
It will discover all unindexed nodes within the range, logging the details with logger `inteligr8.asie.reconcile`.
Optionally, an ASIE `REINDEX` command may be issued against ASIE.</p>
<p>The following path parameters are supported:</p>
<dl>
<dt>fromDbId</dt>
<dd>A DB ID integer for the starting point of a range, inclusive.</dd>
<dt>toDbId</dt>
<dd>A DB ID integer for the ending point of a range, exclusive.</dd>
</dl>
<p>The following response body should be expected in most cases (200, 202, and 500 status codes):</p>
<pre>
{
"unreconciled": [
number, // node DB ID
...
],
"success": {
"number": [ // node DB ID
"solrHostSync:8983/solr",
...
],
...
},
"scheduled": {
"number": [ // node DB ID
"solrHostAsync:8983/solr",
...
],
...
},
"error": {
"number": [ // node DB ID
"solrHostThatFailed:8983/solr": {
"message": "string"
},
...
],
...
}
}
</pre>
<p>The following status codes should be expected:</p>
<dl>
<dt>202</dt>
<dd>Accepted</dd>
<dt>400</dt>
<dd>The path or query parameters are invalid</dd>
</dl>
]]></description>
<!-- Endpoint Configuration -->
<url>/inteligr8/asie/acs/nodes/{fromDbId}/{toDbId}/reconcile?reindex={reindex?}</url>
<format default="json">any</format>
<!-- Security -->
<authentication>user</authentication>
<!-- Functionality -->
<cache>
<never>false</never>
<public>false</public>
</cache>
</webscript>

View File

@@ -21,7 +21,10 @@
<url>/inteligr8/asie/nodes</url>
<!-- Security -->
<authentication>admin</authentication>
<authentication>user</authentication>
<!-- Transaction -->
<transaction>required</transaction>
<!-- Functionality -->
<cache>

View File

@@ -0,0 +1,63 @@
<webscript xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:noNamespaceSchemaLocation="https://bitbucket.org/!api/2.0/snippets/inteligr8/AzMgbp/80fdd26a6b3769a63cdc6b54bf1f39e378545cf7/files/snippet.txt">
<!-- Naming & Organization -->
<shortname>Reindex ACS Node in ASIE Indexes</shortname>
<family>Inteligr8 ASIE</family>
<description><![CDATA[
<p>Reindex the specified ACS node in the ASIE indexes.
This call will attempt to reindex the ACS node on all applicable Solr nodes.
The reindex operation could be synchronous or asynchronous and could fail on any Solr node.
If any Solr node failed synchronously in the execution, then expect a status code of 500.
The response body will still be identical to the 200/202 status codes.
If any Solr node is executing the reindex asynchronously and there are no synchronous failures, then expect a status code of 202.</p>
<p>The following path parameters are supported:</p>
<dl>
<dt>nodeId</dt>
<dd>An ACS node ID.</dd>
</dl>
<p>The following response body should be expected in most cases (200, 202, and 500 status codes):</p>
<pre>
{
"nodeDbId": number,
"success": [
"solrHostSync:8983/solr",
...
],
"scheduled": [
"solrHostAsync:8983/solr",
...
],
"error": [
"solrHostThatFailed:8983/solr": {
"message": "string"
},
...
]
}
</pre>
<p>The following status codes should be expected:</p>
<dl>
<dt>200</dt>
<dd>OK</dd>
<dt>202</dt>
<dd>Accepted</dd>
<dt>400</dt>
<dd>The path or query parameters are invalid</dd>
</dl>
]]></description>
<!-- Endpoint Configuration -->
<url>/inteligr8/asie/acs/node/{nodeId}/reindex</url>
<format default="json">any</format>
<!-- Security -->
<authentication>user</authentication>
<!-- Functionality -->
<cache>
<never>false</never>
<public>false</public>
</cache>
</webscript>

View File

@@ -0,0 +1,46 @@
<webscript xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:noNamespaceSchemaLocation="https://bitbucket.org/!api/2.0/snippets/inteligr8/AzMgbp/80fdd26a6b3769a63cdc6b54bf1f39e378545cf7/files/snippet.txt">
<!-- Naming & Organization -->
<shortname>Retry ASIE Indexes</shortname>
<family>Inteligr8 ASIE</family>
<description><![CDATA[
<p>Issue a 'retry' command to the ASIE indexes.
This call will attempt to retry all remembered failed ACS nodes on Solr nodes.
The retry operation is asynchronous and could fail on any Solr node without notification.</p>
<p>The following response body should be expected in most cases (202 and 500 status codes):</p>
<pre>
{
"scheduled": [
"solrHostAsync:8983/solr",
...
],
"error": [
"solrHostThatFailed:8983/solr": {
"message": "string"
},
...
]
}
</pre>
<p>The following status codes should be expected:</p>
<dl>
<dt>202</dt>
<dd>Accepted</dd>
</dl>
]]></description>
<!-- Endpoint Configuration -->
<url>/inteligr8/asie/acs/retry</url>
<format default="json">any</format>
<!-- Security -->
<authentication>user</authentication>
<!-- Functionality -->
<cache>
<never>false</never>
<public>false</public>
</cache>
</webscript>

View File

@@ -47,7 +47,7 @@
<format default="json">any</format>
<!-- Security -->
<authentication>admin</authentication>
<authentication>user</authentication>
<!-- Functionality -->
<cache>

View File

@@ -60,7 +60,7 @@
<format default="json">any</format>
<!-- Security -->
<authentication>admin</authentication>
<authentication>user</authentication>
<!-- Functionality -->
<cache>

View File

@@ -57,7 +57,7 @@
<format default="json">any</format>
<!-- Security -->
<authentication>admin</authentication>
<authentication>user</authentication>
<!-- Functionality -->
<cache>

View File

@@ -2,7 +2,7 @@
# defaulting to 3 days = 60 * 24 * 3 = 4320
inteligr8.asie.backup.persistTimeMinutes=4320
inteligr8.asie.allowedAuthorities=ALFRESCO_ADMINISTRATORS
inteligr8.asie.allowedAuthorities=GROUP_ALFRESCO_ADMINISTRATORS
# same as solr.baseUrl, but that property is private to the Search subsystem
inteligr8.asie.basePath=/solr

View File

@@ -1,3 +1,6 @@
logger.inteligr8-asie.name=com.inteligr8.alfresco.asie
logger.inteligr8-asie.level=INFO
logger.reconcile.name=inteligr8.asie.reconcile
logger.reconcile.level=INFO

View File

@@ -0,0 +1,92 @@
package com.inteligr8.alfresco.asie.util;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.apache.commons.lang3.mutable.MutableInt;
import org.apache.commons.lang3.tuple.Pair;
import org.junit.Assert;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class CompositeFutureUnitTest {
private final Logger logger = LoggerFactory.getLogger(this.getClass());
@Test
public void test() throws InterruptedException, ExecutionException, TimeoutException {
ExecutorService executor = Executors.newWorkStealingPool(3);
CompositeFuture<Void> futures = new CompositeFuture<Void>();
this.logger.trace("Execution submitting");
futures.combine(executor.submit(new Pausable(100L)));
this.logger.trace("Execution submitting");
futures.combine(executor.submit(new Pausable(200L)));
this.logger.trace("Execution submitting");
futures.combine(executor.submit(new Pausable(300L)));
this.logger.trace("Execution submitting");
futures.combine(executor.submit(new Pausable(400L)));
this.logger.trace("Execution submitting");
futures.combine(executor.submit(new Pausable(500L)));
Thread.sleep(50L);
Assert.assertEquals(Pair.of(5, 0), futures.counts());
Thread.sleep(100L);
Assert.assertEquals(Pair.of(4, 1), futures.counts());
Thread.sleep(100L);
Assert.assertEquals(Pair.of(3, 2), futures.counts());
Thread.sleep(100L);
Assert.assertEquals(Pair.of(2, 3), futures.counts());
Thread.sleep(100L);
Assert.assertEquals(Pair.of(2, 3), futures.counts());
Thread.sleep(100L);
Assert.assertEquals(Pair.of(1, 4), futures.counts());
Thread.sleep(200L);
Assert.assertEquals(Pair.of(0, 5), futures.counts());
long startTime = System.currentTimeMillis();
futures.get();
Assert.assertTrue(System.currentTimeMillis() - startTime < 5);
startTime = System.currentTimeMillis();
futures.get(10L, TimeUnit.SECONDS);
Assert.assertTrue(System.currentTimeMillis() - startTime < 5);
}
private class Pausable implements Callable<Void> {
private static final MutableInt SEQUENCE = new MutableInt(1);
private final int seq;
private final long pauseInMillis;
public Pausable(long pauseInMillis) {
synchronized (SEQUENCE) {
this.seq = SEQUENCE.getAndIncrement();
}
this.pauseInMillis = pauseInMillis;
}
@Override
public Void call() throws InterruptedException {
logger.trace("Thread started execution: " + Thread.currentThread().getName() + ": " + this.seq);
Thread.sleep(this.pauseInMillis);
logger.trace("Thread ending execution: " + Thread.currentThread().getName() + ": " + this.seq);
return null;
}
}
}

View File

@@ -0,0 +1,146 @@
package com.inteligr8.alfresco.asie.util;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.apache.commons.lang3.mutable.MutableInt;
import org.apache.commons.lang3.tuple.Pair;
import org.junit.Assert;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class ThrottledThreadPoolExecutorUnitTest {
private final Logger logger = LoggerFactory.getLogger(this.getClass());
@Test(expected = RejectedExecutionException.class)
public void testHardRejection() throws InterruptedException, ExecutionException, TimeoutException {
ThrottledThreadPoolExecutor executor = new ThrottledThreadPoolExecutor(1, 1, 1, 10L, TimeUnit.SECONDS, "not-throttled");
try {
long startTime = System.currentTimeMillis();
CompositeFuture<Void> futures = new CompositeFuture<Void>();
this.logger.trace("Execution submitting");
futures.combine(executor.submit(new Pausable(100L)));
// confirm there is no blocking as the pool is filled
this.assertElapsed(startTime, 0L, 50L);
Assert.assertEquals(Pair.of(1, 0), futures.counts());
this.logger.trace("Execution submitting");
futures.combine(executor.submit(new Pausable(200L)));
// confirm there is no blocking as the queue is filled
this.assertElapsed(startTime, 0L, 50L);
Assert.assertEquals(Pair.of(2, 0), futures.counts());
this.logger.trace("Execution submitting");
futures.combine(executor.submit(new Pausable(300L)));
} finally {
executor.shutdown();
Assert.assertTrue(executor.awaitTermination(2L, TimeUnit.SECONDS));
}
}
@Test
public void testBlocking() throws InterruptedException, ExecutionException, TimeoutException {
ThrottledThreadPoolExecutor executor = new ThrottledThreadPoolExecutor(2, 2, 2, 10L, TimeUnit.SECONDS, "throttled");
try {
long startTime = System.currentTimeMillis();
CompositeFuture<Void> futures = new CompositeFuture<Void>();
this.logger.trace("Execution submitting");
futures.combine(executor.submit(new Pausable(100L), 1L, TimeUnit.SECONDS));
this.logger.trace("Execution submitting");
futures.combine(executor.submit(new Pausable(200L), 1L, TimeUnit.SECONDS));
// confirm there is no blocking as the pool is filled
this.assertElapsed(startTime, 0L, 50L);
Assert.assertEquals(Pair.of(2, 0), futures.counts());
this.logger.trace("Execution submitting");
futures.combine(executor.submit(new Pausable(300L), 1L, TimeUnit.SECONDS));
this.logger.trace("Execution submitting");
futures.combine(executor.submit(new Pausable(400L), 1L, TimeUnit.SECONDS));
// confirm there is no blocking as the queue is filled
this.assertElapsed(startTime, 0L, 50L);
Assert.assertEquals(Pair.of(4, 0), futures.counts());
this.logger.trace("Execution submitting");
futures.combine(executor.submit(new Pausable(500L), 10L, TimeUnit.SECONDS));
// confirm there is blocking as the pool and queue are full
this.assertElapsed(startTime, 100L, 150L);
Assert.assertEquals(Pair.of(4, 1), futures.counts());
executor.shutdown();
Thread.sleep(150L); // 250ms
Assert.assertEquals(Pair.of(3, 2), futures.counts());
Thread.sleep(100L); // 350ms
Assert.assertEquals(Pair.of(3, 2), futures.counts());
Thread.sleep(100L); // 450ms
Assert.assertEquals(Pair.of(2, 3), futures.counts());
Thread.sleep(100L); // 550ms
Assert.assertEquals(Pair.of(2, 3), futures.counts());
Thread.sleep(100L); // 650ms
Assert.assertEquals(Pair.of(1, 4), futures.counts());
Thread.sleep(200L); // 850ms
Assert.assertEquals(Pair.of(1, 4), futures.counts());
Thread.sleep(100L); // 950 ms
Assert.assertEquals(Pair.of(0, 5), futures.counts());
startTime = System.currentTimeMillis();
futures.get();
this.assertElapsed(startTime, 0L, 25L);
startTime = System.currentTimeMillis();
futures.get(10L, TimeUnit.SECONDS);
this.assertElapsed(startTime, 0L, 25L);
} finally {
executor.awaitTermination(2L, TimeUnit.SECONDS);
}
}
private void assertElapsed(long startTime, long minRangeInclusive, long maxRangeExclusive) {
long elapsedTime = System.currentTimeMillis() - startTime;
Assert.assertTrue("The execution time was shorter than expected: " + elapsedTime + " ms < " + minRangeInclusive + " ms", elapsedTime >= minRangeInclusive);
Assert.assertTrue("The execution time was longer than expected: " + elapsedTime + " ms >= " + maxRangeExclusive + " ms", elapsedTime < maxRangeExclusive);
}
private class Pausable implements Callable<Void> {
private static final MutableInt SEQUENCE = new MutableInt(1);
private final int seq;
private final long pauseInMillis;
public Pausable(long pauseInMillis) {
synchronized (SEQUENCE) {
this.seq = SEQUENCE.getAndIncrement();
}
this.pauseInMillis = pauseInMillis;
}
@Override
public Void call() throws InterruptedException {
logger.trace("Thread started execution: " + Thread.currentThread().getName() + ": " + this.seq);
Thread.sleep(this.pauseInMillis);
logger.trace("Thread ending execution: " + Thread.currentThread().getName() + ": " + this.seq);
return null;
}
}
}

View File

@@ -0,0 +1,10 @@
rootLogger.level=trace
rootLogger.appenderRef.stdout.ref=STDOUT
logger.this.name=com.inteligr8.alfresco.asie
logger.this.level=trace
appender.stdout.type=Console
appender.stdout.name=STDOUT
appender.stdout.layout.type=PatternLayout
appender.stdout.layout.pattern=%d{ABSOLUTE_MICROS} %level{length=1} %c{1}: %m%n

View File

@@ -6,16 +6,16 @@
<parent>
<groupId>com.inteligr8.alfresco</groupId>
<artifactId>asie-platform-module-parent</artifactId>
<version>1.2.1</version>
<version>1.2.2</version>
<relativePath>../</relativePath>
</parent>
<groupId>com.inteligr8</groupId>
<artifactId>solr-api</artifactId>
<version>1.0.0-solr6</version>
<version>1.1.0-solr6</version>
<packaging>jar</packaging>
<name>Apache Solr JAX-RS API</name>
<name>Apache Solr Jakarta RS API</name>
<properties>
<jackson.version>2.18.0</jackson.version>

View File

@@ -1,7 +1,7 @@
package com.inteligr8.solr.api;
import com.inteligr8.solr.model.ActionResponse;
import com.inteligr8.solr.model.ResponseAction;
import com.inteligr8.solr.model.Action;
import com.inteligr8.solr.model.collection.AliasesResponse;
import com.inteligr8.solr.model.collection.GetAliasesRequest;
import com.inteligr8.solr.model.core.ReloadRequest;
@@ -17,7 +17,7 @@ public interface CollectionAdminApi {
@GET
@Produces(MediaType.APPLICATION_JSON)
ActionResponse<ResponseAction> reload(@BeanParam ReloadRequest request);
ActionResponse<Action> reload(@BeanParam ReloadRequest request);
@GET
@Produces(MediaType.APPLICATION_JSON)

View File

@@ -2,10 +2,9 @@ package com.inteligr8.solr.model;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.annotation.JsonProperty.Access;
@JsonIgnoreProperties(ignoreUnknown = true)
public class ResponseAction {
public class Action {
public enum Status {
@JsonProperty("success")
@@ -16,26 +15,18 @@ public class ResponseAction {
Error,
}
@JsonProperty(access = Access.READ_ONLY)
@JsonProperty
private Status status;
@JsonProperty(access = Access.READ_ONLY)
@JsonProperty
private String errorMessage;
public Status getStatus() {
return status;
}
protected void setStatus(Status status) {
this.status = status;
}
public String getErrorMessage() {
return errorMessage;
}
protected void setErrorMessage(String errorMessage) {
this.errorMessage = errorMessage;
}
}

View File

@@ -2,12 +2,11 @@ package com.inteligr8.solr.model;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.annotation.JsonProperty.Access;
@JsonIgnoreProperties(ignoreUnknown = true)
public class ActionResponse<T extends ResponseAction> extends BaseResponse {
public class ActionResponse<T extends Action> extends BaseResponse {
@JsonProperty(access = Access.READ_ONLY)
@JsonProperty
private T action;
public T getAction() {

View File

@@ -2,20 +2,15 @@ package com.inteligr8.solr.model;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.annotation.JsonProperty.Access;
@JsonIgnoreProperties(ignoreUnknown = true)
public class BaseResponse {
@JsonProperty(access = Access.READ_ONLY)
@JsonProperty(required = true)
private ResponseHeader responseHeader;
public ResponseHeader getResponseHeader() {
return responseHeader;
}
protected void setResponseHeader(ResponseHeader responseHeader) {
this.responseHeader = responseHeader;
}
}

View File

@@ -0,0 +1,23 @@
package com.inteligr8.solr.model;
import java.util.HashMap;
import java.util.Map;
import com.fasterxml.jackson.annotation.JsonAnySetter;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
@JsonIgnoreProperties(ignoreUnknown = true)
public class Cores<T> {
private Map<String, T> cores = new HashMap<>();
public T getByCore(String core) {
return cores.get(core);
}
@JsonAnySetter
public void addCore(String core, T value) {
this.cores.put(core, value);
}
}

View File

@@ -0,0 +1,27 @@
package com.inteligr8.solr.model;
import java.util.HashMap;
import java.util.Map;
import com.fasterxml.jackson.annotation.JsonAnySetter;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
@JsonIgnoreProperties(ignoreUnknown = true)
public class Metadata {
private Map<String, Object> metadata = new HashMap<>();
public Map<String, Object> getAll() {
return this.metadata;
}
public Object getByField(String field) {
return this.metadata.get(field);
}
@JsonAnySetter
public void setMetadata(String field, Object value) {
this.metadata.put(field, value);
}
}

View File

@@ -2,31 +2,22 @@ package com.inteligr8.solr.model;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.annotation.JsonProperty.Access;
@JsonIgnoreProperties(ignoreUnknown = true)
public class ResponseHeader {
@JsonProperty(value = "QTime", access = Access.READ_ONLY)
@JsonProperty(value = "QTime", required = true)
private long executionTimeInMilliseconds;
@JsonProperty(access = Access.READ_ONLY)
@JsonProperty(required = true)
private int status;
public long getExecutionTimeInMilliseconds() {
return executionTimeInMilliseconds;
}
protected void setExecutionTimeInMilliseconds(long executionTimeInMilliseconds) {
this.executionTimeInMilliseconds = executionTimeInMilliseconds;
}
public int getStatus() {
return status;
}
protected void setStatus(int status) {
this.status = status;
}
}

Some files were not shown because too many files have changed in this diff Show More