Mercurial Hosting > nabble

comparison src/nabble/model/CursorNodeIterator.java @ 0:7ecd1a4ef557

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
add content
author Franklin Schmidt <fschmidt@gmail.com>
date Thu, 21 Mar 2019 19:15:52 -0600
parents
children
comparison
equal deleted inserted replaced
-1:000000000000 0:7ecd1a4ef557
1 package nabble.model;
2
3 import java.sql.Connection;
4 import java.sql.Statement;
5 import java.sql.PreparedStatement;
6 import java.sql.ResultSet;
7 import java.sql.SQLException;
8 import java.util.List;
9 import java.util.ArrayList;
10 import java.util.NoSuchElementException;
11 import java.util.concurrent.atomic.AtomicInteger;
12 import org.slf4j.Logger;
13 import org.slf4j.LoggerFactory;
14
15
16 final class CursorNodeIterator extends NodeIterator<NodeImpl> {
17 private static final Logger logger = LoggerFactory.getLogger(CursorNodeIterator.class);
18
19 private static final int fetchSize = Init.get("fetchSize",100);
20
21 private final SiteKey siteKey;
22 private final Connection con;
23 private final PreparedStatement stmt;
24 private boolean isClosed = false;
25 private ResultSet rs = null;
26 private int offset = 0;
27 private boolean hasCheckedNext = false;
28 private NodeImpl next = null;
29 private final boolean oldAutoCommit;
30 private static final AtomicInteger counter = new AtomicInteger();
31 private final String cursor = "crs" + counter.incrementAndGet();
32 // private final Exception initException = new Exception("init");
33
34 CursorNodeIterator(SiteKey siteKey,String sql,DbParamSetter paramSetter) {
35 try {
36 this.siteKey = siteKey;
37 this.con = siteKey.getDb().getConnection();
38 PreparedStatement pstmt = con.prepareStatement(
39 "declare " + cursor + " cursor with hold for " +sql
40 );
41 Connection pgCon = pstmt.getConnection();
42 oldAutoCommit = pgCon.getAutoCommit();
43 if( oldAutoCommit )
44 pgCon.setAutoCommit(false);
45 paramSetter.setParams(pstmt);
46 pstmt.execute();
47 pstmt.close();
48 this.stmt = con.prepareStatement(
49 "fetch " + fetchSize + " " + cursor
50 );
51 } catch(SQLException e) {
52 close();
53 throw new RuntimeException(sql,e);
54 } catch(RuntimeException e) {
55 close();
56 throw e;
57 }
58 }
59
60 private void openRs()
61 throws SQLException
62 {
63 rs = stmt.executeQuery();
64 }
65
66 private void closeRs()
67 throws SQLException
68 {
69 rs.close();
70 rs = null;
71 }
72
73 @Override public boolean hasNext() {
74 if( isClosed )
75 return false;
76 if( !hasCheckedNext ) {
77 hasCheckedNext = true;
78 try {
79 if( rs == null )
80 openRs();
81 if( rs.next() ) {
82 next = NodeImpl.getNode(siteKey,rs);
83 if( ++offset % fetchSize == 0 )
84 closeRs();
85 } else {
86 next = null;
87 close();
88 }
89 } catch(SQLException e) {
90 throw new RuntimeException(e);
91 }
92 }
93 return next != null;
94 }
95
96 @Override public NodeImpl next() {
97 if( !hasNext() )
98 throw new NoSuchElementException();
99 hasCheckedNext = false;
100 return next;
101 }
102
103 @Override public void skip(int n) {
104 try {
105 if( n==0 )
106 return;
107 if( hasCheckedNext ) {
108 hasCheckedNext = false;
109 if( --n == 0 )
110 return;
111 }
112 if( rs != null ) {
113 int left = fetchSize - (offset % fetchSize);
114 if( n < left ) {
115 rs.relative(n);
116 return;
117 }
118 closeRs();
119 offset += left;
120 n -= left;
121 }
122 if( n >= fetchSize ) {
123 int fwd = n / fetchSize * fetchSize;
124 n -= fwd;
125 offset += fwd;
126 Statement stmt = con.createStatement();
127 stmt.execute(
128 "move " + fwd + " " + cursor
129 );
130 stmt.close();
131 }
132 if( n == 0 )
133 return;
134 openRs();
135 rs.relative(n);
136 offset += n;
137 } catch(SQLException e) {
138 throw new RuntimeException("n = "+n,e);
139 }
140 }
141
142 @Override public List<Node> get(int i,int n) {
143 try {
144 i += offset;
145 Statement stmt = con.createStatement();
146 stmt.execute(
147 "move absolute " + i + " " + cursor
148 );
149 ResultSet rs = stmt.executeQuery(
150 "fetch " + n + " " + cursor
151 );
152 List<Node> list = new ArrayList<Node>();
153 while( rs.next() ) {
154 list.add( NodeImpl.getNode(siteKey,rs) );
155 }
156 rs.close();
157 stmt.close();
158 close();
159 return list;
160 } catch(SQLException e) {
161 throw new RuntimeException(e);
162 }
163 }
164
165 @Override public List<NodeImpl> asList() {
166 try {
167 List<NodeImpl> list = new ArrayList<NodeImpl>();
168 if( rs != null ) {
169 while( rs.next() ) {
170 list.add( NodeImpl.getNode(siteKey,rs) );
171 }
172 closeRs();
173 }
174 Statement stmt = con.createStatement();
175 ResultSet rs = stmt.executeQuery(
176 "fetch forward all " + cursor
177 );
178 while( rs.next() ) {
179 list.add( NodeImpl.getNode(siteKey,rs) );
180 }
181 rs.close();
182 stmt.close();
183 close();
184 return list;
185 } catch(SQLException e) {
186 throw new RuntimeException(e);
187 }
188 }
189
190 @Override public void close() {
191 if( isClosed )
192 return;
193 isClosed = true;
194 try {
195 if( rs != null )
196 rs.close();
197 if( stmt != null ) {
198 Connection pgCon = stmt.getConnection();
199 stmt.close();
200 {
201 Statement stmt = con.createStatement();
202 stmt.execute(
203 "close " + cursor
204 );
205 stmt.close();
206 }
207 if( oldAutoCommit )
208 pgCon.setAutoCommit(true);
209 }
210 if( con != null )
211 con.close();
212 } catch(SQLException e) {
213 throw new RuntimeException(e);
214 }
215 }
216
217 @Override protected void finalize() throws Throwable {
218 if( !isClosed ) {
219 logger.error("didn't close NodeIterator"/*,initException*/);
220 close();
221 }
222 super.finalize();
223 }
224
225 }