Mercurial Hosting > luan
annotate src/goodjava/lucene/backup/BackupIndexWriter.java @ 1738:9713f7fd50b3
server-sent events
| author | Franklin Schmidt <fschmidt@gmail.com> |
|---|---|
| date | Thu, 03 Nov 2022 19:23:53 -0600 |
| parents | 973d3039c421 |
| children | 1578324d2aac |
| rev | line source |
|---|---|
| 1488 | 1 package goodjava.lucene.backup; |
| 2 | |
| 3 import java.io.File; | |
| 1499 | 4 import java.io.InputStream; |
| 1674 | 5 import java.io.OutputStream; |
| 6 import java.io.BufferedOutputStream; | |
| 7 import java.io.FileOutputStream; | |
| 1488 | 8 import java.io.IOException; |
| 1499 | 9 import java.net.Socket; |
| 1672 | 10 import java.net.ConnectException; |
| 1488 | 11 import java.util.List; |
| 12 import java.util.ArrayList; | |
| 1499 | 13 import java.util.Map; |
| 14 import java.util.HashMap; | |
| 15 import java.util.Arrays; | |
| 1504 | 16 import java.util.concurrent.Executors; |
| 17 import java.util.concurrent.ExecutorService; | |
| 1672 | 18 import java.util.concurrent.ExecutionException; |
|
1508
86c5e7000ecf
lucene.backup checksum
Franklin Schmidt <fschmidt@gmail.com>
parents:
1506
diff
changeset
|
19 import org.apache.lucene.search.SortField; |
| 1488 | 20 import goodjava.io.IoUtils; |
| 1499 | 21 import goodjava.rpc.RpcClient; |
| 22 import goodjava.rpc.RpcCall; | |
| 23 import goodjava.rpc.RpcResult; | |
| 24 import goodjava.rpc.RpcException; | |
| 1488 | 25 import goodjava.lucene.api.LuceneIndexWriter; |
| 26 import goodjava.lucene.logging.LoggingIndexWriter; | |
| 27 import goodjava.lucene.logging.LogFile; | |
| 1672 | 28 import goodjava.logging.Logger; |
| 29 import goodjava.logging.LoggerFactory; | |
| 1488 | 30 |
| 31 | |
| 1672 | 32 public final class BackupIndexWriter extends LoggingIndexWriter { |
| 1499 | 33 private static final Logger logger = LoggerFactory.getLogger(BackupIndexWriter.class); |
| 34 public static String[] backupDomains; | |
|
1690
973d3039c421
backup server checks client domain
Franklin Schmidt <fschmidt@gmail.com>
parents:
1675
diff
changeset
|
35 private final String domain; |
| 1488 | 36 private final String name; |
| 37 private final File dir; | |
| 1499 | 38 private boolean isSyncPending = false; |
| 1504 | 39 private final ExecutorService exec = Executors.newSingleThreadExecutor(); |
| 1488 | 40 |
|
1690
973d3039c421
backup server checks client domain
Franklin Schmidt <fschmidt@gmail.com>
parents:
1675
diff
changeset
|
41 public BackupIndexWriter(LuceneIndexWriter indexWriter,File logDir,long logTime,String domain,String name) |
| 1548 | 42 throws IOException |
| 43 { | |
| 44 super(indexWriter,logDir,logTime); | |
| 1499 | 45 if( backupDomains == null ) |
| 46 throw new RuntimeException("must set backupDomains"); | |
|
1690
973d3039c421
backup server checks client domain
Franklin Schmidt <fschmidt@gmail.com>
parents:
1675
diff
changeset
|
47 this.domain = domain; |
| 1488 | 48 this.name = name; |
| 49 File f = new File(System.getProperty("java.io.tmpdir")); | |
|
1690
973d3039c421
backup server checks client domain
Franklin Schmidt <fschmidt@gmail.com>
parents:
1675
diff
changeset
|
50 dir = new File(f,"goodjava.lucene/"+domain+"~"+name); |
| 1501 | 51 IoUtils.mkdirs(dir); |
| 1488 | 52 } |
| 53 | |
| 1672 | 54 @Override public synchronized void close() throws IOException { |
| 1504 | 55 super.close(); |
| 56 exec.shutdown(); | |
| 57 } | |
| 58 | |
| 1672 | 59 @Override public synchronized void commit() throws IOException { |
| 1488 | 60 super.commit(); |
| 1499 | 61 //sync(); |
| 62 if( !isSyncPending ) { | |
| 1504 | 63 exec.execute(sync); |
| 1499 | 64 isSyncPending = true; |
| 1488 | 65 } |
| 66 } | |
| 67 | |
| 1672 | 68 @Override protected boolean doCheck(SortField sortField) throws IOException { |
|
1538
634f6765830e
use goodjava/lucene/logging
Franklin Schmidt <fschmidt@gmail.com>
parents:
1512
diff
changeset
|
69 boolean ok = super.doCheck(sortField); |
|
634f6765830e
use goodjava/lucene/logging
Franklin Schmidt <fschmidt@gmail.com>
parents:
1512
diff
changeset
|
70 if( ok ) |
|
634f6765830e
use goodjava/lucene/logging
Franklin Schmidt <fschmidt@gmail.com>
parents:
1512
diff
changeset
|
71 runSyncWithChecksum(); |
|
634f6765830e
use goodjava/lucene/logging
Franklin Schmidt <fschmidt@gmail.com>
parents:
1512
diff
changeset
|
72 return ok; |
|
1508
86c5e7000ecf
lucene.backup checksum
Franklin Schmidt <fschmidt@gmail.com>
parents:
1506
diff
changeset
|
73 } |
|
86c5e7000ecf
lucene.backup checksum
Franklin Schmidt <fschmidt@gmail.com>
parents:
1506
diff
changeset
|
74 |
| 1499 | 75 public void runSync() { |
| 1672 | 76 exec(sync); |
| 1499 | 77 } |
| 78 | |
|
1508
86c5e7000ecf
lucene.backup checksum
Franklin Schmidt <fschmidt@gmail.com>
parents:
1506
diff
changeset
|
79 public void runSyncWithChecksum() { |
| 1672 | 80 exec(syncWithChecksum); |
| 81 } | |
| 82 | |
| 83 private void exec(Runnable r) { | |
|
1508
86c5e7000ecf
lucene.backup checksum
Franklin Schmidt <fschmidt@gmail.com>
parents:
1506
diff
changeset
|
84 try { |
| 1672 | 85 exec.submit(r).get(); |
| 86 } catch(InterruptedException e) { | |
| 87 throw new RuntimeException(e); | |
| 88 } catch(ExecutionException e) { | |
| 89 Throwable cause = e.getCause(); | |
| 90 if( cause instanceof RuntimeException ) | |
| 91 throw (RuntimeException)cause; | |
|
1508
86c5e7000ecf
lucene.backup checksum
Franklin Schmidt <fschmidt@gmail.com>
parents:
1506
diff
changeset
|
92 throw new RuntimeException(e); |
|
86c5e7000ecf
lucene.backup checksum
Franklin Schmidt <fschmidt@gmail.com>
parents:
1506
diff
changeset
|
93 } |
|
86c5e7000ecf
lucene.backup checksum
Franklin Schmidt <fschmidt@gmail.com>
parents:
1506
diff
changeset
|
94 } |
|
86c5e7000ecf
lucene.backup checksum
Franklin Schmidt <fschmidt@gmail.com>
parents:
1506
diff
changeset
|
95 |
| 1499 | 96 private final Runnable sync = new Runnable() { |
| 1504 | 97 public void run() { |
| 1499 | 98 try { |
|
1508
86c5e7000ecf
lucene.backup checksum
Franklin Schmidt <fschmidt@gmail.com>
parents:
1506
diff
changeset
|
99 sync(false); |
| 1672 | 100 } catch(ConnectException e) { |
| 101 logger.error("sync failed: "+e.getMessage()); | |
| 1499 | 102 } catch(IOException e) { |
| 103 throw new RuntimeException(e); | |
| 104 } | |
| 105 } | |
| 106 }; | |
| 107 | |
|
1508
86c5e7000ecf
lucene.backup checksum
Franklin Schmidt <fschmidt@gmail.com>
parents:
1506
diff
changeset
|
108 private final Runnable syncWithChecksum = new Runnable() { |
|
86c5e7000ecf
lucene.backup checksum
Franklin Schmidt <fschmidt@gmail.com>
parents:
1506
diff
changeset
|
109 public void run() { |
|
86c5e7000ecf
lucene.backup checksum
Franklin Schmidt <fschmidt@gmail.com>
parents:
1506
diff
changeset
|
110 try { |
|
86c5e7000ecf
lucene.backup checksum
Franklin Schmidt <fschmidt@gmail.com>
parents:
1506
diff
changeset
|
111 sync(true); |
| 1672 | 112 } catch(ConnectException e) { |
| 113 logger.error("syncWithChecksum failed: "+e.getMessage()); | |
|
1508
86c5e7000ecf
lucene.backup checksum
Franklin Schmidt <fschmidt@gmail.com>
parents:
1506
diff
changeset
|
114 } catch(IOException e) { |
|
86c5e7000ecf
lucene.backup checksum
Franklin Schmidt <fschmidt@gmail.com>
parents:
1506
diff
changeset
|
115 throw new RuntimeException(e); |
|
86c5e7000ecf
lucene.backup checksum
Franklin Schmidt <fschmidt@gmail.com>
parents:
1506
diff
changeset
|
116 } |
|
86c5e7000ecf
lucene.backup checksum
Franklin Schmidt <fschmidt@gmail.com>
parents:
1506
diff
changeset
|
117 } |
|
86c5e7000ecf
lucene.backup checksum
Franklin Schmidt <fschmidt@gmail.com>
parents:
1506
diff
changeset
|
118 }; |
|
86c5e7000ecf
lucene.backup checksum
Franklin Schmidt <fschmidt@gmail.com>
parents:
1506
diff
changeset
|
119 |
|
86c5e7000ecf
lucene.backup checksum
Franklin Schmidt <fschmidt@gmail.com>
parents:
1506
diff
changeset
|
120 private void sync(boolean withChecksum) throws IOException { |
| 1499 | 121 List<LogFile> logs = new ArrayList<LogFile>(); |
| 122 synchronized(this) { | |
| 123 isSyncPending = false; | |
| 1500 | 124 clearDir(); |
| 1499 | 125 for( LogFile log : this.logs ) { |
| 126 File f = new File(dir,log.file.getName()); | |
| 127 IoUtils.link(log.file,f); | |
| 128 logs.add( new LogFile(f) ); | |
| 129 } | |
| 130 } | |
| 131 List logInfo = new ArrayList(); | |
| 132 Map<String,LogFile> logMap = new HashMap<String,LogFile>(); | |
| 133 for( LogFile log : logs ) { | |
| 134 Map fileInfo = new HashMap(); | |
| 135 fileInfo.put("name",log.file.getName()); | |
| 136 fileInfo.put("end",log.end()); | |
|
1508
86c5e7000ecf
lucene.backup checksum
Franklin Schmidt <fschmidt@gmail.com>
parents:
1506
diff
changeset
|
137 if( withChecksum ) |
|
86c5e7000ecf
lucene.backup checksum
Franklin Schmidt <fschmidt@gmail.com>
parents:
1506
diff
changeset
|
138 fileInfo.put("checksum",log.checksum()); |
| 1499 | 139 logInfo.add(fileInfo); |
| 140 logMap.put(log.file.getName(),log); | |
| 141 } | |
| 142 for( String backupDomain : backupDomains ) { | |
| 1509 | 143 RpcClient rpc = BackupServer.rpcClient(backupDomain); |
| 1499 | 144 try { |
|
1690
973d3039c421
backup server checks client domain
Franklin Schmidt <fschmidt@gmail.com>
parents:
1675
diff
changeset
|
145 RpcCall call = new RpcCall("login",domain,name); |
| 1512 | 146 rpc.write(call); |
| 147 rpc.read(); | |
| 148 call = new RpcCall("check",logInfo); | |
| 1499 | 149 while(true) { |
| 150 rpc.write(call); | |
| 151 RpcResult result = rpc.read(); | |
| 1512 | 152 //logger.info(Arrays.asList(result.returnValues).toString()); |
| 1499 | 153 String status = (String)result.returnValues[0]; |
| 154 if( status.equals("ok") ) { | |
| 155 break; | |
|
1508
86c5e7000ecf
lucene.backup checksum
Franklin Schmidt <fschmidt@gmail.com>
parents:
1506
diff
changeset
|
156 } else if( status.equals("missing") || status.equals("bad_checksum") ) { |
| 1499 | 157 String fileName = (String)result.returnValues[1]; |
|
1508
86c5e7000ecf
lucene.backup checksum
Franklin Schmidt <fschmidt@gmail.com>
parents:
1506
diff
changeset
|
158 if( status.equals("bad_checksum") ) |
|
86c5e7000ecf
lucene.backup checksum
Franklin Schmidt <fschmidt@gmail.com>
parents:
1506
diff
changeset
|
159 logger.error("bad_checksum "+fileName); |
| 1499 | 160 LogFile log = logMap.get(fileName); |
| 161 long len = log.end() - 8; | |
| 162 InputStream in = log.input(); | |
| 1512 | 163 call = new RpcCall(in,len,"add",logInfo,fileName); |
| 1675 | 164 logger.info("add "+fileName); |
| 1499 | 165 } else if( status.equals("incomplete") ) { |
| 166 String fileName = (String)result.returnValues[1]; | |
| 167 long logEnd = (Long)result.returnValues[2]; | |
| 168 LogFile log = logMap.get(fileName); | |
| 169 long len = log.end() - logEnd; | |
| 170 InputStream in = log.input(); | |
| 171 in.skip(logEnd-8); | |
| 1512 | 172 call = new RpcCall(in,len,"append",logInfo,fileName); |
| 1675 | 173 // logger.info("append "+fileName); |
| 1499 | 174 } else |
| 175 throw new RuntimeException("status "+status); | |
| 176 } | |
| 177 } catch(RpcException e) { | |
| 1672 | 178 logger.error("",e); |
| 1499 | 179 } |
| 180 rpc.close(); | |
| 181 } | |
| 1500 | 182 clearDir(); |
| 183 } | |
| 184 | |
| 185 private void clearDir() throws IOException { | |
| 186 for( File f : dir.listFiles() ) { | |
| 187 IoUtils.delete(f); | |
| 188 } | |
| 1499 | 189 } |
| 190 | |
|
1690
973d3039c421
backup server checks client domain
Franklin Schmidt <fschmidt@gmail.com>
parents:
1675
diff
changeset
|
191 public static BackupIndexWriter newWithRestore(LuceneIndexWriter indexWriter,File logDir,long logTime,String domain,String name) |
| 1674 | 192 throws IOException |
| 193 { | |
| 194 if( !logDir.exists() ) { | |
| 195 RpcClient rpc = BackupServer.rpcClient(backupDomains[0]); | |
| 196 try { | |
| 197 RpcCall call; | |
| 198 RpcResult result; | |
|
1690
973d3039c421
backup server checks client domain
Franklin Schmidt <fschmidt@gmail.com>
parents:
1675
diff
changeset
|
199 call = new RpcCall("exists",domain,name); |
| 1674 | 200 rpc.write(call); |
| 201 result = rpc.read(); | |
| 202 boolean exists = (Boolean)result.returnValues[0]; | |
| 203 if( exists ) { | |
| 204 logger.error("restoring "+logDir+" from backup"); | |
| 205 File zip = File.createTempFile("luan_",".zip"); | |
| 206 IoUtils.delete(zip); | |
|
1690
973d3039c421
backup server checks client domain
Franklin Schmidt <fschmidt@gmail.com>
parents:
1675
diff
changeset
|
207 call = new RpcCall("login",domain,name); |
| 1674 | 208 rpc.write(call); |
| 209 rpc.read(); | |
| 210 call = new RpcCall("zip"); | |
| 211 rpc.write(call); | |
| 212 result = rpc.read(); | |
| 213 OutputStream out = new BufferedOutputStream(new FileOutputStream(zip)); | |
| 214 IoUtils.copyAll(result.in,out); | |
| 215 out.close(); | |
| 216 IoUtils.mkdirs(logDir); | |
| 217 String cmd = "unzip " + zip; | |
| 218 Process proc = Runtime.getRuntime().exec(cmd,null,logDir); | |
| 219 IoUtils.waitFor(proc); | |
| 220 IoUtils.delete(zip); | |
| 221 } | |
| 222 } catch(RpcException e) { | |
| 223 throw new RuntimeException(e); | |
| 224 } | |
| 225 rpc.close(); | |
| 226 } | |
|
1690
973d3039c421
backup server checks client domain
Franklin Schmidt <fschmidt@gmail.com>
parents:
1675
diff
changeset
|
227 return new BackupIndexWriter(indexWriter,logDir,logTime,domain,name); |
| 1674 | 228 } |
| 229 | |
| 1488 | 230 } |