1: <?php
  2: namespace phpcassa\Iterator;
  3: 
  4: use phpcassa\ColumnFamily;
  5: 
  6: /**
  7:  * @package phpcassa/Iterator
  8:  */
  9: class ColumnFamilyIterator implements \Iterator {
 10: 
 11:     protected $column_family;
 12:     protected $buffer_size;
 13:     protected $row_count;
 14:     protected $read_consistency_level;
 15:     protected $column_parent, $predicate;
 16: 
 17:     protected $current_buffer, $buffer_number;
 18:     protected $next_start_key, $orig_start_key;
 19:     protected $is_valid;
 20:     protected $rows_seen;
 21: 
 22:     protected function __construct($column_family,
 23:                                    $buffer_size,
 24:                                    $row_count,
 25:                                    $orig_start_key,
 26:                                    $column_parent,
 27:                                    $predicate,
 28:                                    $read_consistency_level) {
 29: 
 30:         $this->column_family = $column_family;
 31:         $this->buffer_size = $buffer_size;
 32:         $this->row_count = $row_count;
 33:         $this->orig_start_key = $orig_start_key;
 34:         $this->next_start_key = $orig_start_key;
 35:         $this->column_parent = $column_parent;
 36:         $this->predicate = $predicate;
 37:         $this->read_consistency_level = $read_consistency_level;
 38:         $this->array_format = $column_family->return_format == ColumnFamily::ARRAY_FORMAT;
 39: 
 40:         if ($this->row_count !== null)
 41:             $this->buffer_size = min($this->row_count, $buffer_size);
 42: 
 43:         $this->buffer_number = 0;
 44:     }
 45: 
 46:     public function rewind() {
 47:         // Setup first buffer
 48:         $this->rows_seen = 0;
 49:         $this->is_valid = true;
 50:         $this->next_start_key = $this->orig_start_key;
 51:         $this->get_buffer();
 52: 
 53:         # If nothing was inserted, this may happen
 54:         if (count($this->current_buffer) == 0) {
 55:             $this->is_valid = false;
 56:             return;
 57:         }
 58: 
 59:         # If the very first row is a deleted row
 60:         if (count(current($this->current_buffer)) == 0)
 61:             $this->next();
 62:         else
 63:             $this->rows_seen++;
 64:     }
 65: 
 66:     public function current() {
 67:         return current($this->current_buffer);
 68:     }
 69: 
 70:     public function key() {
 71:         return key($this->current_buffer);
 72:     }
 73: 
 74:     public function valid() {
 75:         return $this->is_valid;
 76:     }
 77: 
 78:     public function next() {
 79:         if ($this->rows_seen === $this->row_count) {
 80:             $this->is_valid = false;
 81:             return;
 82:         }
 83: 
 84:         $beyond_last_row = false;
 85:         # If we haven't run off the end
 86:         if ($this->current_buffer != null)
 87:         {
 88:             # Save this key in case we run off the end
 89:             if ($this->array_format) {
 90:                 $cur = current($this->current_buffer);
 91:                 $this->next_start_key = $cur[0];
 92:             } else {
 93:                 $this->next_start_key = key($this->current_buffer);
 94:             }
 95:             next($this->current_buffer);
 96: 
 97:             if (count(current($this->current_buffer)) == 0)
 98:             {
 99:                 # this is an empty row, skip it
100:                 do {
101:                     $this->next_start_key = key($this->current_buffer);
102:                     next($this->current_buffer);
103:                     $key = key($this->current_buffer);
104:                     if ( !isset($key) ) {
105:                         $beyond_last_row = true;
106:                         break;
107:                     }
108:                 } while (count(current($this->current_buffer)) == 0);
109:             }
110:             else
111:             {
112:                 $key = key($this->current_buffer);
113:                 $beyond_last_row = !isset($key);
114:             }
115: 
116:             if (!$beyond_last_row)
117:             {
118:                 $this->rows_seen++;
119:                 if ($this->rows_seen > $this->row_count) {
120:                     $this->is_valid = false;
121:                     return;
122:                 }
123:             }
124:         }
125:         else
126:         {
127:             $beyond_last_row = true;
128:         }
129: 
130:         if($beyond_last_row && $this->current_page_size < $this->expected_page_size)
131:         {
132:             # The page was shorter than we expected, so we know that this
133:             # was the last page in the column family
134:             $this->is_valid = false;
135:         }
136:         else if($beyond_last_row)
137:         {
138:             # We've reached the end of this page, but there should be more
139:             # in the CF
140: 
141:             # Get the next buffer (next_start_key has already been set)
142:             # we loop through multiple calls of get_buffer() in case the entire
143:             # buffer load is tombstones, the old method of just calling next() could
144:             # result in excessive recursion
145:             while ( $this->is_valid ) {
146:                 $this->get_buffer();
147: 
148:                 # If the result set is 1, we can stop because the first item
149:                 # should always be skipped if we're not on the first buffer
150:                 if(count($this->current_buffer) == 1 && $this->buffer_number != 1)
151:                 {
152:                     $this->is_valid = false;
153:                     break;
154:                 }
155:                 else
156:                 {
157:                     // skip leading tombstone rows
158:                     $need_new_buffer = false;
159:                     $skipped_a_row = false;
160:                     while (count(current($this->current_buffer)) == 0) {
161:                         $skipped_a_row = true;
162: 
163:                         $this->next_start_key = key($this->current_buffer);
164:                         next($this->current_buffer);
165:                         $key = key($this->current_buffer);
166: 
167:                         if ( !isset($key) ) {
168:                             if ( $this->current_page_size < $this->expected_page_size ) {
169:                                 // looks like we're at the last page of data so just give up
170:                                 $this->is_valid = false;
171:                             } else {
172:                                 // found nothing but tombstones, fetch a new buffer
173:                                 $need_new_buffer = true;
174:                             }
175:                             break;
176:                         }
177:                     }
178: 
179:                     if ( !$need_new_buffer ) {
180:                         if (!$skipped_a_row) {
181:                             // The first row in the new buffer needs to be skipped
182:                             $this->next();
183:                         } else {
184:                             // We already took care of skipping the first row
185:                             // because it happened to be a tombstone. We're now
186:                             // pointing at a valid row, so increment our seen
187:                             // row count.
188:                             $this->rows_seen++;
189:                             if ($this->rows_seen > $this->row_count) {
190:                                 $this->is_valid = false;
191:                                 return;
192:                             }
193:                         }
194:                         break;
195:                     }
196:                 }
197:             }
198:         }
199:     }
200: }
201: