Author: sergiykarpenko Date: 2010-11-24 06:03:28 -0500 (Wed, 24 Nov 2010) New Revision: 3531 Modified: jcr/trunk/exo.jcr.component.ext/src/main/java/org/exoplatform/services/jcr/ext/repository/creation/RepositoryCreationServiceImpl.java Log: EXOJCR-929: RepositoryCreationServiceImpl RPC support added Modified: jcr/trunk/exo.jcr.component.ext/src/main/java/org/exoplatform/services/jcr/ext/repository/creation/RepositoryCreationServiceImpl.java =================================================================== --- jcr/trunk/exo.jcr.component.ext/src/main/java/org/exoplatform/services/jcr/ext/repository/creation/RepositoryCreationServiceImpl.java 2010-11-24 09:56:16 UTC (rev 3530) +++ jcr/trunk/exo.jcr.component.ext/src/main/java/org/exoplatform/services/jcr/ext/repository/creation/RepositoryCreationServiceImpl.java 2010-11-24 11:03:28 UTC (rev 3531) @@ -27,15 +27,29 @@ import org.exoplatform.services.jcr.ext.backup.BackupManager; import org.exoplatform.services.jcr.ext.backup.BackupOperationException; import org.exoplatform.services.jcr.ext.backup.RepositoryBackupChainLog; +import org.exoplatform.services.jcr.impl.Constants; import org.exoplatform.services.jcr.impl.storage.jdbc.JDBCWorkspaceDataContainer; import org.exoplatform.services.log.ExoLogger; import org.exoplatform.services.log.Log; import org.exoplatform.services.naming.InitialContextInitializer; +import org.exoplatform.services.rpc.RPCException; import org.exoplatform.services.rpc.RPCService; +import org.exoplatform.services.rpc.RemoteCommand; +import org.exoplatform.ws.frameworks.json.JsonHandler; +import org.exoplatform.ws.frameworks.json.JsonParser; +import org.exoplatform.ws.frameworks.json.impl.BeanBuilder; +import org.exoplatform.ws.frameworks.json.impl.JsonDefaultHandler; +import org.exoplatform.ws.frameworks.json.impl.JsonException; +import org.exoplatform.ws.frameworks.json.impl.JsonGeneratorImpl; +import org.exoplatform.ws.frameworks.json.impl.JsonParserImpl; +import org.exoplatform.ws.frameworks.json.value.JsonValue; import org.jboss.cache.util.concurrent.ConcurrentHashSet; +import java.io.ByteArrayInputStream; import java.io.File; import java.io.FileNotFoundException; +import java.io.InputStream; +import java.io.Serializable; import java.util.HashMap; import java.util.HashSet; import java.util.Map; @@ -64,20 +78,36 @@ private final RepositoryService repositoryService; /** - * The RPC Service used to communicate with other nodes + * The RPC Service used to communicate with other nodes. */ private final RPCService rpcService; + /** + * BackupManager used to restore repository from backup. + */ private final BackupManager backupManager; + /** + * DBCreator used to create database. Only database not tables, indexes, etc. + */ private final DBCreator dbCreator; + /** + * InitalContextInitalizer used to bind new datasource. + */ private final InitialContextInitializer initialContextInitializer; + /** + * Store of reserved repository names. + */ private final Set<String> pendingRepositories = new ConcurrentHashSet<String>(); - private final Set<String> namesUnderNegotiation = new ConcurrentHashSet<String>(); + private RemoteCommand reserveRepositoryName; + private RemoteCommand createRepository; + + private RemoteCommand startRepository; + public RepositoryCreationServiceImpl(RepositoryService repositoryService, BackupManager backupManager, DBCreator dbCreator, InitialContextInitializer initialContextInitializer, final RPCService rpcService) { @@ -86,6 +116,121 @@ this.rpcService = rpcService; this.dbCreator = dbCreator; this.initialContextInitializer = initialContextInitializer; + + if (rpcService == null) + { + LOG.warn("RepositoryCreationService initialized with null RPCService, so other cluser nodes will" + + " not be notified about new repositories."); + } + else + { + // register commands + reserveRepositoryName = rpcService.registerCommand(new RemoteCommand() + { + + public String getId() + { + return "org.exoplatform.services.jcr.ext.repository.creation.RepositoryCreationServiceImpl-reserveRepositoryName"; + } + + public Serializable execute(Serializable[] args) throws Throwable + { + System.out.println("EXECUTED reserveRepositoryName"); + + String repositoryName = (String)args[0]; + if (!pendingRepositories.contains(repositoryName)) + { + pendingRepositories.add(repositoryName); + } + else + { + throw new RepositoryCreationException("Repository name " + repositoryName + " already reserved."); + } + return repositoryName; + } + }); + createRepository = rpcService.registerCommand(new RemoteCommand() + { + + public String getId() + { + return "org.exoplatform.services.jcr.ext.repository.creation.RepositoryCreationServiceImpl-createRepository"; + } + + public Serializable execute(Serializable[] args) throws Throwable + { + System.out.println("EXECUTED createRepository"); + //String backupId, RepositoryEntry rEntry, String rToken + String backupId = (String)args[0]; + String stringRepositoryEntry = (String)args[1]; + String rToken = (String)args[2]; + + try + { + RepositoryEntry rEntry = + (RepositoryEntry)(getObject(RepositoryEntry.class, stringRepositoryEntry + .getBytes(Constants.DEFAULT_ENCODING))); + + createRepo(backupId, rEntry, rToken); + + // execute startRepository at all cluster nodes (coordinator will ignore this command) + try + { + rpcService.executeCommandOnAllNodes(startRepository, false, stringRepositoryEntry); + } + catch (RPCException e) + { + Throwable cause = e.getCause(); + if (cause instanceof RepositoryCreationException) + { + throw new RepositoryCreationException( + "Repository " + rEntry.getName() + + " created on coordinator, can not be started at other cluster node: " + + cause.getMessage(), cause); + } + else + { + throw new RepositoryCreationException("Repository " + rEntry.getName() + + " created on coordinator, can not be started at other cluster node: " + e.getMessage(), e); + } + } + return null; + } + finally + { + // release tokens + pendingRepositories.remove(rToken); + } + } + }); + + startRepository = rpcService.registerCommand(new RemoteCommand() + { + + public String getId() + { + return "org.exoplatform.services.jcr.ext.repository.creation.RepositoryCreationServiceImpl-startRepository"; + } + + public Serializable execute(Serializable[] args) throws Throwable + { + System.out.println("EXECUTED startRepository"); + // must not be executed on coordinator node, since coordinator node already created the repository + if (!rpcService.isCoordinator()) + { + //RepositoryEntry (as String) rEntry + String stringRepositoryEntry = (String)args[0]; + RepositoryEntry rEntry = + (RepositoryEntry)(getObject(RepositoryEntry.class, stringRepositoryEntry + .getBytes(Constants.DEFAULT_ENCODING))); + + startRepository(rEntry); + } + return null; + } + }); + } + } /** @@ -94,23 +239,126 @@ public void createRepository(String backupId, RepositoryEntry rEntry) throws RepositoryConfigurationException, RepositoryCreationException { - reserveRepositoryName(rEntry.getName()); - // 1. check possibility to create repository locally - // - check existing, pending repository and datasources with same names + String rToken = reserveRepositoryName(rEntry.getName()); + createRepository(backupId, rEntry, rToken); + } - // 2. reserve name and put additional information (ex. ip and port of current machine) - // 3. check possibility to create repository on others nodes - // - sending to all cluster nodes information about new repository and waiting for answers - // - all cluster nodes receive information and check possibility to create repository locally - // - send response - // 4. reserve name on all nodes of cluster + /** + * {@inheritDoc} + */ + public void createRepository(String backupId, RepositoryEntry rEntry, String rToken) + throws RepositoryConfigurationException, RepositoryCreationException + { + if (rpcService != null) + { + //TODO is is correct to use Json code here + String stringRepositoryEntry = null; + try + { + JsonGeneratorImpl generatorImpl = new JsonGeneratorImpl(); + JsonValue json = generatorImpl.createJsonObject(rEntry); + stringRepositoryEntry = json.toString(); + } + catch (JsonException e) + { + throw new RepositoryCreationException("Can not serialize repository entry: " + e.getMessage(), e); + } + + try + { + Object result = + rpcService.executeCommandOnCoordinator(createRepository, true, backupId, stringRepositoryEntry, rToken); + + if (result != null) + { + throw new RepositoryCreationException("ReserveRepositoryName command returns unknown command."); + } + } + catch (RPCException e) + { + Throwable cause = ((RPCException)e).getCause(); + if (cause instanceof RepositoryCreationException) + { + throw (RepositoryCreationException)cause; + } + else if (cause instanceof RepositoryConfigurationException) + { + throw (RepositoryConfigurationException)cause; + } + else + { + throw new RepositoryCreationException(e.getMessage(), e); + } + } + } + else + { + try + { + createRepo(backupId, rEntry, rToken); + } + finally + { + pendingRepositories.remove(rEntry); + } + } } /** * {@inheritDoc} */ - public void createRepository(String backupId, RepositoryEntry rEntry, String rToken) + public String reserveRepositoryName(String repositoryName) throws RepositoryCreationException + { + + // check possibility to create repository locally + //check does repository already created + for (int i = 0; i < repositoryService.getConfig().getRepositoryConfigurations().size(); i++) + { + RepositoryEntry conf = repositoryService.getConfig().getRepositoryConfigurations().get(i); + if (conf.getName().equals(repositoryName)) + { + throw new RepositoryCreationException("Repository " + repositoryName + " already exists."); + } + } + + if (rpcService != null) + { + //ask other nodes does they have pending repositories + try + { + Object result = rpcService.executeCommandOnCoordinator(reserveRepositoryName, true, repositoryName); + + if (result instanceof String) + { + return (String)result; + } + else + { + throw new RepositoryCreationException("ReserveRepositoryName command returns unknown type result."); + } + } + catch (RPCException e) + { + Throwable cause = ((RPCException)e).getCause(); + if (cause instanceof RepositoryCreationException) + { + throw (RepositoryCreationException)cause; + } + else + { + throw new RepositoryCreationException(e.getMessage(), e); + } + } + } + else + { + pendingRepositories.add(repositoryName); + return repositoryName; + } + } + + protected void createRepo(String backupId, RepositoryEntry rEntry, String rToken) throws RepositoryConfigurationException, RepositoryCreationException { // check does token registered @@ -232,66 +480,48 @@ throw new RepositoryCreationException("Backup log file by id " + backupId + (backLog != null ? (" and file path=" + backLog.getAbsolutePath()) : "") + " do not exists."); } - //TODO execute "clone repository" on other cluster nodes - // release tokens } - /** - * {@inheritDoc} - */ - public String reserveRepositoryName(String repositoryName) throws RepositoryCreationException + protected void startRepository(RepositoryEntry repositoryEntry) throws RepositoryCreationException { - // 1. check possibility to create repository locally - // - check existing, pending repository and datasources with same names + //TODO do we need bind datasource here? - for (int i = 0; i < repositoryService.getConfig().getRepositoryConfigurations().size(); i++) + try { - RepositoryEntry conf = repositoryService.getConfig().getRepositoryConfigurations().get(i); - if (conf.getName().equals(repositoryName)) - { - throw new RepositoryCreationException("Repository " + repositoryName + " already exists."); - } - + repositoryService.createRepository(repositoryEntry); } - - // try - // { - // ManageableRepository repo = repositoryService.getRepository(repositoryName); - // if (repo != null) - // { - // throw new RepositoryCreationException("Repository " + repositoryName + " already exists."); - // } - // } - // catch (RepositoryException e) - // { - // throw new RepositoryCreationException(e.getMessage(), e); - // } - // catch (RepositoryConfigurationException e) - // { - // throw new RepositoryCreationException(e.getMessage(), e); - // } - - if (namesUnderNegotiation.contains(repositoryName) || pendingRepositories.contains(repositoryName)) + catch (RepositoryConfigurationException e) { - throw new RepositoryCreationException("Repository name " + repositoryName + " already reserved."); + throw new RepositoryCreationException("Repository created but can not be started on this node: " + + e.getLocalizedMessage(), e); } + catch (RepositoryException e) + { + throw new RepositoryCreationException("Repository created but can not be started on this node: " + + e.getLocalizedMessage(), e); + } + } - // 2. reserve name and put additional information (ex. ip and port of current machine) - namesUnderNegotiation.add(repositoryName); + /** + * TODO make another serialization scheme or move it to dedicated object + * Will be created the Object from JSON binary data. + * + * @param cl + * Class + * @param data + * binary data (JSON) + * @return Object + * @throws Exception + * will be generated Exception + */ + private Object getObject(Class cl, byte[] data) throws Exception + { + JsonHandler jsonHandler = new JsonDefaultHandler(); + JsonParser jsonParser = new JsonParserImpl(); + InputStream inputStream = new ByteArrayInputStream(data); + jsonParser.parse(inputStream, jsonHandler); + JsonValue jsonValue = jsonHandler.getJsonObject(); - // TODO ask other nodes does thay have pending repositories - - pendingRepositories.add(repositoryName); - namesUnderNegotiation.remove(repositoryName); - // TODO register repositoryName on other nodes - - // 3. check possibility to create repository on others nodes - // - sending to all cluster nodes information about new repository and waiting for answers - // - all cluster nodes receive information and check possibility to create repository locally - // - send response - // 4. reserve name on all nodes of cluster - - return repositoryName; + return new BeanBuilder().createObject(cl, jsonValue); } - }